第 35 章 电商商品供给、库存、审核与运营全生命周期设计

本章定位：这不是对“商品中心”“供给平台”“库存系统”三篇文章的简单拼接，而是站在平台生命周期视角，把商品如何进入平台、如何被治理、如何形成正式交易契约、如何建立可售库存、如何持续编辑和运营、以及如何与搜索、营销、订单、履约保持一致，收敛成一条完整主线。

如果前面的专题章节分别回答的是“某个系统内部怎么设计”，这一章回答的则是：

1. 平台到底在处理哪些高频使用场景。
2. 这些场景为什么不能用一个后台 CRUD 搞定。
3. Draft、Staging、正式商品、库存事实、审核状态、发布版本应该分别放在哪里。
4. 为什么库存运营入口和库存事实归属必须分离。
5. 为什么真正难的不是建几张表，而是跨系统一致性、任务化、幂等、补偿和资损防控。

建议配合以下章节交叉阅读：

• 第 25 章 商品中心系统
• 第 27 章 库存系统
• 第 26 章 商品供给、编辑、运营与生命周期治理

1. 核心使用场景：系统到底在处理什么问题

很多团队一开始会从系统模块讲起：商品中心一章、库存系统一章、审核后台一章、供应商同步一章。这样讲虽然清楚，但读者很容易失去真正的主线，因为业务里遇到的从来不是“我现在只在操作商品中心”，而是：

一个商品从创建、审核、发布、补库存、改标题、供应商同步、搜索刷新到订单履约，会跨越多少系统边界，以及这些边界怎么保证不打架。

1.1 商品创建场景

商品创建并不是单一操作，而是至少五类不同来源的组合：

这些场景共同逼着系统回答几个问题：

1. 手工创建和批量导入能否走同一条治理主线。
2. 供应商同步是“创建”还是“Upsert”。
3. 外部来源的数据先放哪里，什么时候才算正式商品。
4. 审核和发布失败后如何定位到具体行、具体字段、具体来源。

但真正落到业务现场时，“创建商品” 往往并不只是在建一条商品记录，而是在同时决定这个商品如何建立可售库存能力。尤其在酒旅、到店餐饮、景区票务和本地生活券场景里，商品创建常常天然伴随着库存创建。

典型地会出现 3 类组合场景：

这意味着系统在创建商品时，至少还要回答两个额外问题：

1. 创建商品时，是否必须同步声明 inventory_mode。
2. 商品创建和库存初始化，是不是要被收敛到同一个供给任务里。

也正因为如此，后面第 3.7 节讨论的“库存入口设计”，并不是独立于商品创建存在的附属能力，而是商品供给主链路的一部分。

1.2 商品编辑场景

商品一旦上线，编辑链路往往比创建链路更复杂，因为它直接影响线上交易和历史一致性。

这里最关键的分歧不是“能不能改”，而是：

• 哪些字段必须先进 Staging 再发布。
• 哪些字段可以走快速旁路。
• 哪些变更必须保留审批与审计链路。
• 供应商同步和人工编辑冲突时谁优先。

1.3 库存运营场景

库存不是一个孤立的“数字服务”，B 端运营视角里它至少包含以下场景：

库存场景的难点不是表单操作，而是：

• 库存运营入口要不要直接放在库存系统。
• 数量库存和券码库存能否共用一个模型。
• Redis 里的热数据是不是事实来源。
• 订单预占、支付确认、超时释放怎么和库存账本对齐。

1.4 生命周期治理场景

治理场景决定了这个系统是不是“平台”：

这部分要解决的核心问题是：流程状态、审核状态和正式商品状态能不能混在一起。答案通常是否定的。

1.5 供应商协同场景

供应商同步既是供给入口，也是持续运营的一部分：

供应商协同最大的工程难点不是“能不能接上接口”，而是：

• 供应商数据是不是可以直接覆盖正式商品。
• 同步失败后如何知道哪一批、哪一页、哪个资源出问题。
• 供应商变更和人工编辑冲突时谁说了算。

1.6 场景到系统问题的映射

2. 整体方案设计

这一节按照“先看系统边界，再看主链路，最后看关键决策”的顺序展开。先把商品主域、供给平台、库存系统、营销、搜索和订单各自的职责划清楚，再把这些系统放进一条完整的生命周期主链路中理解，最后集中讨论 Draft / Staging、同步 / 异步、资产归属和库存边界这些关键设计选择。

2.1 系统的边界和职责

边界划分里最重要的三条原则是：

1. 供给平台负责接脏数据、洗数据、做任务和审核编排，但不持有正式商品资产主权。
2. 商品主域写侧负责可审核、可发布的草稿资产、版本冻结和正式落库闭环。
3. 库存系统负责库存事实和账本，不负责 B 端商品编辑与审核流程。

2.2 全生命周期主链路总览

flowchart LR
    A["供给入口"] --> B["Draft / Staging"]
    B --> C["标准化与校验"]
    C --> D["Diff / 风险识别"]
    D --> E["QC / 自动准入"]
    E --> F["发布到商品主域正式态"]
    F --> G["库存 / 营销 / 搜索 / 计价协同"]
    G --> H["订单使用快照交易"]
    F --> I["Outbox 下游投影"]
    I --> J["巡检 / 对账 / 补偿"]

这条主链路表达的是统一供给治理平台的基本职责：

1. 所有供给动作先进入 Draft / Staging，而不是直接污染正式商品。
2. 标准化、校验、Diff、风险识别和审核，构成正式发布前的质量门禁。
3. 发布之后不是“流程结束”，而是进入库存、营销、搜索、计价和订单快照的协同阶段。
4. 最终通过 Outbox、巡检、对账和补偿保证一致性，而不是要求一次同步调用把所有下游都写成功。

2.3 决策点 1：为什么不能继续在原商品系统上打补丁，而是需要新增 Draft 表

场景与问题

很多团队在系统演进早期都会问一个问题：既然原来已经有商品系统，为什么不继续在原表上加几个字段、补几个状态、加几段审核逻辑，而是要显式引入 Draft 表甚至新的写模型？

短期当然可以继续打补丁，但问题是业务模型已经变了。早期很多商品系统主要承载的是自运营的相对简单的配置型商品，商品本身更像一个后台配置对象；后面商品开始承载外部供给接入、审核流程、库存策略、履约规则、退款口径、供应商同步和版本化发布之后，它已经不再只是“静态配置”，而是在向“可治理的交易资产”演进。

如果继续把这些新能力都硬塞进原商品系统的单一正式表或单一状态字段里，短期看似节省改造成本，长期却会持续累积三类问题：

1. 边界继续塌陷

   原商品系统原本只负责正式商品表达，如果继续往里塞草稿、审核、导入状态、供应商同步中间态，它就会同时承担接入层、流程层和资产层职责，后面每接一个新品类或一个新供给来源，复杂度都会继续上升。

2. 正式资产被流程态污染

   一旦把“待审、驳回、文件解析中、待补图、供应商刷新中”这些状态混进正式商品表，商品主域的语义就不再稳定，搜索、缓存、计价、订单也更容易误读这些半成品或中间态。

3. 演进成本越来越高

   继续打补丁的本质是拿局部 if/else 和临时字段去扛新的业务模型。前期似乎更快，但后期每增加一种商品类型、每新增一种审核策略、每接一种库存模式，都要在旧结构上继续叠复杂度，风险和维护成本只会越来越高。

所以这次改造的本质不是简单重构代码，而是边界重划：把正式商品、可发布草稿、流程任务、审核工作台、库存事实这些不同层次的数据重新分层。也正因为如此，系统才需要显式引入 Draft 或 Staging 这类新写模型，而不是继续在原商品表上追加字段和状态。

方案对比

推荐方案

推荐方案 B。判断标准不是“新增表是不是更优雅”，而是业务模型是否已经发生质变。

一旦商品开始承载供给接入、审核治理、库存协同、履约规则和版本化发布，原商品系统就不再适合只靠补字段和补状态来演进。此时新增 Draft 表，本质上是在承认“正式商品”和“待治理草稿”是两种不同层次的数据，应该被显式建模。

2.4 决策点 2：Draft 属于供给平台还是商品主域写侧，QC 挂在哪层快照上

场景与问题

这是整个商品生命周期架构里最容易引发争论的一个决策点。按业务流程直觉看，商家和运营是在供给平台改商品，所以草稿似乎天然应该留在供给侧；但从资产主权、事务边界、版本冻结和长期演进看，Draft 又更像商品主域内部的未生效版本。

这个问题不能只看“草稿表放哪里”，还必须连着 QC 链路一起看。因为真正的架构问题不是：

Draft 在供给还是在商品中心？

而是：

可审核、可发布的草稿资产归谁持有，QC 又应该挂在哪一层冻结快照上做裁判？

如果 Draft 留在供给侧，QC 往往也只能围绕供给侧草稿运行，链路就会自然变成：

供给(draft)
  → QC
  → 商品正式态

如果 Draft 已进入商品主域写侧，QC 更合理的形态则是：

供给接入
  → 商品主域 Draft / Staging
  → QC
  → 本地 Merge 到正式 Item

这两条链路看起来只差了一个步骤，底层却是两种完全不同的资产主权和版本控制模型。

这里先给出结论，再展开论证：

如果 Draft 只是浏览器表单保存、文件上传中的临时缓存，它可以短暂停留在供给侧；但一旦 Draft 进入待审核、待发布、可版本冻结、可与正式商品合流的阶段，它本质上已经属于商品主域的写模型，而不应该长期留在供给接入层。

换句话说，这一节讨论的不是“前端临时草稿存哪里”，而是“可审核、可发布的商品草稿资产归谁持有”。

方案 A：Draft 留在供给平台

这种方案的出发点非常自然：供给平台负责对接商家、运营、供应商和 ERP，商品变更从这里进入，审核工作台、批量任务、错误文件也在这里，于是团队很容易顺着流程心智把 Draft 一起留在供给层。

在这种设计下，QC 也通常会变成“前置流式审核”，即先在供给侧审核，再把通过后的 DTO 推给商品主域落正式态。

优点

• 业务流程看起来更顺，商家上传、审核流、任务进度都集中在一个平台。
• 供给平台天然适合承接脏数据、错误文件、批量导入和人工修复。
• 对早期团队来说，上线速度快，不需要商品主域一开始就承接完整的草稿写模型。

缺点 / 风险

1. 资产主权分裂

   同一个商品会被拆成两份核心表达：供给平台里的草稿资产，以及商品中心里的正式资产。草稿和正式商品其实只是同一资产的不同阶段，却被两个服务分别持有，后续版本控制、冲突治理和审计解释都会变复杂。

2. 发布合流链路跨服务，放大批量窗口风险

   当 QC 判定通过时，系统需要把草稿 merge 到正式商品。如果草稿在供给、正式表在商品中心，发布就必须走跨服务 RPC 或异步命令。单次发布未必是灾难，但一旦进入凌晨供应商大批量更新、大促前批量调价、酒店政策全量同步等场景，发布链路时延、锁竞争、失败点和重试成本都会明显放大，也更容易在批量窗口冲击商品主域写链路。

3. 类目属性契约和校验逻辑容易进入“双写地狱”

   商品草稿模型与正式模型往往高度同构：类目、属性、多规格 SKU、阶梯价、履约规则、退款规则都要表达。商品主域一旦新增类目契约、修改必填属性或调整校验规则，供给平台就很容易被迫同步修改草稿表结构和复制一套校验逻辑。久而久之，两个服务会演变成披着接口外衣的伪单体。

4. 版本冻结与防篡改更难做

   如果审核看的数据在供给侧，而最终发布动作发生在商品主域，审核中途还要面对商家、运营、供应商同步并发修改同一商品的问题。理论上可以通过版本号、快照和锁机制解决，但复杂度明显高于把可发布草稿直接内聚在商品主域本地。

5. QC 更难锁定真正的裁判快照

   这是这条链路最微妙、也是最危险的问题。QC 在供给侧看到的是版本 A，但在审核通过到跨服务发布的窗口里，供给侧草稿可能已经被另一个并发修改线程推进到版本 B。最终就可能出现“审的是 A，发的是 B”的时空撕裂问题。

6. 多版本乱序容易导致线上数据回滚

   商家短时间连续提交多个版本时，机审秒过的版本 B 可能先发布，而人工审核较慢的版本 A 后通过。如果发布动作只是“审核通过后把 DTO 再推给商品主域”，旧版本 A 就可能在更晚的时间反向覆盖新版本 B，形成线上数据时间倒流。

适用场景

• 早期系统
• 类目简单、审核弱、供应商同步不重
• 商品主域还没有成熟的写模型和发布版本体系

方案 B：Draft 收拢到商品主域写侧

这里的“放在商品中心”需要说得更精确一些：不是把 Draft 直接放进 C 端线上 product_item 表，而是放进商品主域自己的 draft / staging / version / publish merge 写模型中。商品主域对这份草稿资产拥有版本冻结、发布合流和正式落库的主权。

在这种设计下，QC 不再审核供给平台里那份可变草稿，而是基于商品主域已经落库、已经冻结版本号的 Draft / Staging 快照做裁判。QC 回传的也不再是完整商品 DTO，而只是轻量的判决结果，例如：

{
  "draft_id": 555,
  "publish_version": 102,
  "result": "PASS"
}

优点

1. 资产主权统一

   无论是线上售卖商品，还是待审核的未生效版本，本质上都是同一份商品资产的不同形态。把它们收在商品主域写侧，领域边界最稳定。

2. 合流可以本地事务闭环

   product_draft_tab、product_staging_tab、product_item_tab、publish_version 都在商品主域本地时，QC 通过后的 merge 可以在本地事务内完成。这样发布路径更短，也更容易做版本校验、幂等发布和失败回滚。

3. 版本冻结与快照锁定更自然

   供给平台把数据清洗成标准 DTO 后推给商品主域，商品主域将其落成 Draft 或 Staging 并生成版本号。QC 审核的是这份冻结后的本地快照，而不是仍然暴露在接入层并发修改风险中的动态数据。

4. 契约和校验逻辑只维护一份

   类目、属性、SKU、Offer、履约、退款等商品主契约只需在一个主域里演进，避免供给和商品两边长期维护两套同构模型。

5. 发布事件与读模型刷新更容易统一

   正式商品落库、快照生成、Outbox 写入、搜索/缓存/计价/营销投影刷新都可以由商品主域统一发起，链路更可解释。

6. QC 能基于冻结快照做真正的版本裁判

   数据一旦进入商品主域落成 Draft 或 Staging，系统就可以为它分配明确的业务版本号，并在送审期间锁定这份快照。商家若继续修改，必须生成新的版本而不是覆写旧版本。这样 QC 审核的是确定版本，发布的也是同一个确定版本。

7. QC 回传 verdict，商品主域本地事务完成 merge

   QC 审核通过后，不需要再跨服务传一份巨大的商品 DTO 回商品主域，只需要回传“哪一个草稿版本审核通过”的轻量判决。商品主域随后在本地事务里将该版本 merge 到正式 Item，并统一完成快照、Outbox、缓存刷新等后续动作。

缺点 / 风险

1. B 端写压力会向商品主域集中

   零散保存、批量导入、供应商同步、错误重提等写流量都会压到商品主域写侧。

2. 商品主域会承接更多流程复杂度

   如果团队边界没划清，商品主域很容易被误用成“供给后台本体”，把本该属于供给平台的任务、审核工作台、文件处理逻辑全堆进来。

化解施策

方案 B 的关键不是“把所有东西都塞进一个商品服务”，而是：

• 供给平台继续负责入口、清洗、任务、审核工作台、供应商同步编排。
• 商品主域只负责可审核、可发布的草稿资产，以及正式落库闭环。
• product_draft_tab 与 product_item_tab 物理隔离，避免草稿写洪峰污染线上正式资产。
• 条件成熟时，可以拆成 product-writer 和 product-reader：前者承接草稿、审核、发布；后者只服务 C 端读流量和缓存。

适用场景

• 中大型平台
• 类目复杂、供应商同步重、审核和版本治理要求高
• 希望长期保持商品主权统一的系统

方案对比

推荐方案

推荐方案 B，但需要准确理解为：

Draft 不应该长期留在供给接入层，也不应该直接混进 C 端线上商品表，而应该归商品主域写侧管理。

供给平台的定位更像“流量入口、数据加工厂与流程编排层”：

• 负责接外部脏数据
• 负责把数据清洗成标准 DTO
• 负责任务、审核工作台、错误文件和供应商同步流程

而商品主域才真正负责：

• Draft / Staging / Version
• 版本冻结与防篡改
• QC 裁判快照锚定
• 本地事务 merge
• 正式商品落库
• Outbox 与下游投影事件

因此，更完整的架构判断不是“Draft 属于供给平台还是商品中心”二选一，而是：

• 入口态临时数据 可以短暂停在供给侧
• 可审核、可发布的草稿资产 应归商品主域写侧
• QC 应该裁判商品主域中已经冻结版本的快照

这也是为什么在平台化电商架构里，真正成熟的设计通常都不会让供给平台长期持有商品草稿的最终主权。

2.5 决策点 3：是否需要 Staging，单 Draft + Operation Log 是否可行

场景与问题

当团队决定引入 Draft 之后，下一步一定会遇到两个高度关联的问题：

1. 只有 Draft 和正式 Item 两层够不够，是否还需要单独的 Staging。
2. 如果想尽量轻量化，单 Draft + Operation Log 能不能替代 Draft + Staging + Item 三层模型。

这两个问题最好放在一起讨论，因为它们本质上都在回答同一件事：

商品写模型到底需不需要三层隔离，以及单草稿方案的边界在哪里。

这里先给出一个压缩结论：

• Draft 解决的是“脏数据吸收、审核沙盒、并行编辑”。
• Staging 解决的是“已通过 QC 的干净版本如何等待生效、切渠道、做回滚”。
• Operation Log 适合作为单草稿方案的差分增强，但它不能完全替代 Staging 在时间轴、渠道隔离和回滚上的作用。

Draft 与 Staging 的职责差异

如果没有 Staging，系统就只能在“审核通过后立刻发布”和“继续停留在 Draft 里等待后续动作”之间二选一，这会让定时生效、多渠道未来版本和回滚变得很别扭。

方案 A：只有 Draft，无 Staging

优点

• 模型简单，表和状态少。
• 审核通过后直接发布，链路短。
• 对小系统或低变更频率系统来说，心智负担低。

缺点 / 风险

1. 已提交版本和正在编辑版本容易混淆

   如果没有 Staging，系统很难清晰表达“这份数据已经通过 QC，但暂时还不应该生效”的状态。

2. 定时生效和渠道隔离能力弱

   大促零点变价、不同渠道未来版本、供应商已过审但待外部确认的场景，都需要一个“已过审但未激活”的物理缓冲区。

3. 回滚确定性弱

   没有 Staging 的物理快照，回滚通常要依赖历史版本重算、日志回放或脚本修复。

适用场景

• 小团队
• 商品结构简单
• 几乎没有定时生效、多渠道隔离和秒级回滚要求

方案 B：单 Draft + Operation Log

这是很多中型系统会选择的折中方案：一个商品只保留一个 Draft，再配套 Operation Log / Diff Map 记录价格、标题、库存等字段差分。

优点

• 比三层模型更轻量，只需要 Item + Draft + Operation Log。
• 某些低风险变更可以直接 patch 当前草稿，避免“一个商品只能改一次”的业务阻塞。
• 存储成本、索引成本和模型心智都比较低。

缺点 / 风险

1. 多任务审核容易撕裂

   标题改动需要人工审核 3 小时，价格改动 100ms 秒过，如果它们都 patch 到同一个 Draft 上，价格就无法安全独立发布。

2. 激活时需要动态归并日志

   没有 Staging 的完整待发布快照，系统只能在生效瞬间做 Operation Log 的 Reduce 和字段级 patch，这会把计算压力带到高峰时刻。

3. 多渠道隔离与一键回滚能力弱

   单 Draft 很难同时表达多个未来渠道版本；一旦线上数据洗脏，回滚通常要依赖逆向回放 Operation Log，确定性弱于直接切回上一个 Staging 版本。

适用场景

• 中等复杂度系统
• 审核策略比较一致
• 定时生效和多渠道未来版本需求不强

方案 C：Draft + Staging + Item

优点

• Draft 负责吸收脏写流量和并行编辑，Staging 负责沉淀通过 QC 的完整待发布版本。
• 多个 Draft 可以并行加工，但最终只在 Staging 中形成有序的未来版本时间轴。
• 到激活时不再做复杂归并，而是把已经组装好的快照本地 merge 到 Item。
• 支持定时发布、多渠道隔离、灰度切换和秒级回滚。

缺点 / 风险

• 物理数据副本更多，模型更重。
• 需要额外的激活调度、版本治理和清理策略。
• 对团队的数据建模要求更高。

适用场景

• 中大型平台
• 审核存在快慢混合链路
• 大促定时变价、跨渠道供给隔离较多
• 需要高确定性的回滚和版本切换

方案对比

推荐方案

推荐方案 C 作为复杂供给中台的目标形态，但不是所有系统第一天就必须上三层模型。

更务实的判断标准是：

• 如果系统当前仍以单渠道、弱审核、低频定时发布为主，方案 B 完全可以作为高性价比方案。
• 一旦系统开始出现多路并行修改、价格和图文审核时效分裂、大促零点定时生效、跨渠道未来版本和秒级回滚要求，就应从单 Draft 进化到 Draft + Staging + Item。

换句话说，Staging 不是“为了优雅而优雅”的抽象，而是当系统需要时间轴、渠道隔离和高确定性回滚时，最自然的一层写模型。

2.6 决策点 4：同步体验和异步任务如何分层

推荐方案是方案 C：表单手工创建、低风险单品编辑保留同步交互；批量导入、批量编辑、券码导入、供应商同步全部任务化。

2.7 决策点 5：单商品创建也需要写入任务表吗

场景与问题

这个问题在很多团队里都会被直觉性地处理成“单品同步接口就直接写数据库，不要引入任务表”，因为从前端交互看，运营或商家只是在页面上点一次提交，系统秒级返回，看起来不像一个需要任务化治理的场景。

但如果把链路往后多看几步，就会发现单商品创建后面仍然会经过：

• 标准化与校验
• 交易契约检查
• 风险识别
• 送商品主域写侧冻结快照
• 审核或自动准入
• 发布编排
• 下游刷新
• 失败补偿与审计追踪

如果单品场景完全不落 task，它就会和 Excel、API/ISV 推送、批量编辑形成两套完全不同的排障、审计和补偿口径。

方案对比

推荐方案

推荐方案 C。单商品创建虽然是同步交互，但后端仍建议写入一条轻量 product_supply_task，并在需要时配一条 task_item。这样可以统一：

• 入口受理号 receipt_id
• 操作链路 operation_id
• 发布记录 publish_record
• 失败补偿入口
• 运营侧任务查询与审计口径

这条任务记录不是为了把单品交互“强行异步化”，而是为了让同步体验和长链路治理共用同一套后端锚点。

一句话说，就是：

单商品创建可以同步执行，但不应该脱离任务模型。

2.8 决策点 6：库存配置与库存事实是否放在同一系统

场景与问题

商品上线前常常需要配置库存策略，例如是数量库存、券码库存还是供应商库存；但上线后真正的库存余额、预占和账本又属于交易硬约束。

方案对比

推荐方案

推荐方案 B：库存配置是交易契约的一部分，可以在商品发布时一并声明；库存余额、预占、确认、释放、券码状态机必须由库存系统维护。

2.9 决策点 7：供应商同步的无效流量过滤，应该放在供给侧还是商品中心

场景与问题

供应商同步场景里，经常会遇到一个非常诱人的优化点：既然 100w 级对象里可能有 70%~80% 根本没有实质变化，那么是不是应该在供给侧先基于业务指纹做一轮粗过滤，把无效数据直接蒸发在上游？

这个思路在算力和带宽上很有吸引力，但它会立即引出一个更危险的问题：

供给侧是否真的拥有足够权威的“真相”，来判断一条供应商数据应该被跳过？

如果供给侧维护了一套自己的指纹或状态镜像，而商品中心又是正式商品的唯一真相源，那么一旦出现：

• 上一轮同步失败
• 消息在 MQ 中长时间排队
• 运营在商品中心人工改价或锁定字段
• 多条链路同时修改同一商品

供给侧就很容易基于一份“过期状态”做出错误过滤，最终把本应进入商品中心重试或更新的数据静默跳过，形成最危险的“丢单式不一致”。

方案对比

很多团队在这里还会想到一个“折中优化”：

• 在 supplier_sync_state_ledger 中增加 hash_sign
• Consumer 先把本次供应商对象的核心字段做 MD5
• 如果和 state_ledger.hash_sign 完全一致，就直接在供给侧把对象切成 SKIPPED
• 不再请求商品中心

这个方案看上去能进一步削减商品中心压力，但它本质上仍属于“由供给侧模拟商品真相”的思路，因此需要单独评估：

因此，对于 hash_sign 这类设计，书里建议明确区分 “观测字段” 和 “裁决字段”：

• 可以把 hash_sign 作为一种辅助观测字段，用来统计“本批次疑似无变化对象占比”、辅助排障和大盘分析
• 但不建议由供给侧据此直接裁决 SKIPPED

核心原因很简单：

1. 供给侧的 hash_sign 不代表正式商品当前真相
   它最多只代表“上一次供给链路认为自己处理过什么”。一旦上次写商品失败、超时未知、被运营锁定、被别的链路改过，这个 hash 就可能已经过期。

2. 供应商视角的“没变”，不等于商品中心视角的“可跳过”
   商品中心还要看：

   • 当前 publish_version
   • 字段主导权
   • 运营锁定状态
   • 是否存在其他来源并发修改
     这些信息都不是供应商同步链路自己能最终裁决的。

3. 上游错误过滤比多打一条消息更危险
   多打一条 MQ、让商品中心多做一次只读反查，最多是多一点算力开销；但如果上游错误地把一条本该重试或更新的数据切成 SKIPPED，它就会变成最难排查的静默丢单。

不过，这并不意味着供给侧就只能做“纯搬运”，一条无效流量都不能拦。更准确的说法是：

供给侧不能做最终业务去重裁决，但可以做安全过滤。

这里的“安全过滤”有一个很严格的边界：它只能基于 供给链路自身的确定性状态，去切掉那些数学上或流程上绝对无效的流量，不能去猜测“商品中心现在是不是没变”。

推荐保留的安全过滤包括：

1. 同一批次内的重复页 / 游标漂移过滤

   • 如果同一个 outer_goods_id 在同一 current_batch_id 下已经进入 PROCESSING / SUCCESS
   • 且本次报文 payload_hash 与刚处理过的内容一致
   • 可以安全短路，避免因为供应商翻页重叠导致重复投递

2. 无映射死链过滤

   • 如果对象状态已经明确是业务阻断态，比如 BLOCKED / UNMAPPED
   • 且错误码是 MAPPING_MISSING / MAPPING_REMOVED
   • 则可以在供给侧直接抑制，不再重复把这类对象送入商品中心
   • 当运营补齐映射时，再通过事件把状态重新拨回 INIT

3. 并发重投 / 幽灵消息过滤

   • 如果对象已经被另一个 Consumer 成功 CAS 强占到 PROCESSING
   • 当前这条消息就是明显的重投或短暂并发踩脚
   • 可以直接丢弃或延迟重试，而不必重复读 HBase 和反查商品中心

4. 已明确成功投递过商品中心的数据的重复过滤

   • 如果在 同一批次 内，该对象已经明确完成过一次成功投递，并且当前重入消息与上一次处理内容一致
   • 则可以在供给侧短路，避免把同一条成功消息重复送到商品中心
   • 这里的关键前提是“已经明确成功投递过商品中心”，而不是仅仅因为本地 hash 看起来没变

所以最终边界应该这样理解：

• 供给侧允许做拓扑级、流程级、安全型过滤
• 商品中心负责最终业务级、版本级、真相级幂等去重

推荐方案

推荐方案 B：供给侧不承担最终过滤主权，由商品中心自己基于正式真相做幂等去重和 Diff 裁剪；供给侧仅保留安全过滤。

这背后的核心判断是：

1. 商品中心才是正式商品的唯一真相源

   供应商同步、运营手工编辑、商家后台修改、营销锁价都可能同时作用于同一商品。只有商品中心知道当前线上正式版本、字段锁定状态和最终主导权，因此也只有它最适合做“是否真要写入”的最终裁决。

2. 上游过滤最怕静默丢单

   一旦供给侧的指纹、状态或失败记录与商品中心不一致，就会出现“上游判定没变，实际下游没成功”的黑洞。这类错误比多发几条 MQ 更危险，因为它既难发现，又直接导致数据永远不同步。

3. 幂等和去重本来就是商品主域必须具备的底线能力

   即使没有供应商 Pull，商品中心也必须面对：

   • 同一消息重投
   • 多链路并发更新
   • 旧版本晚到
   • 运营人工修改和自动同步打架

   所以把去重主权收拢到商品中心，不是额外负担，而是让商品主域回到它本来就该承担的职责。

设计收束

这条决策最终会把供应商同步链路压缩成一种更稳定的非对称结构：

• 供给侧：负责串行拉取、快照落盘、状态台账、MQ 投递、任务与批次管理，并只做安全过滤
• 商品中心：负责基于正式 DTO、publish_version、字段主导权和乐观锁做最终幂等拦截

一句话说：

上游允许做拓扑级安全过滤，但最终业务去重不能离开商品中心。
对供应商同步来说，供给侧只能切掉“绝对无效”的流量，最终是否真要跳过，必须在商品主域的最终写入前裁决。

2.10 决策点 8：订单商品快照应该由订单中心在下单时拍，还是由商品中心提前固化

场景与问题

订单要解决的不是“此刻商品线上长什么样”，而是“用户下单那一刻，平台承诺交易的到底是什么”。

这就自然引出一个高频架构问题：

既然订单最终要把商品快照写进 order_item，为什么不让订单中心在创单时自己实时抓商品、自己组装并保存快照？

这个问题表面上看是“少一张表、少一层对象”的简化，但它背后其实是在问：

• 商品结构与交易结构的主权应该落在哪个领域
• 创单链路是否应该承担商品快照组装成本
• 商品快照到底是“交易临时拼装物”，还是“商品主域提前固化的解释事实”

方案对比

推荐方案

推荐方案 B：商品快照由商品中心在发布时提前固化，订单中心在下单时只引用 item_snapshot_id，不自己拼装快照。

核心原因有 3 个：

1. 商品结构主权在商品中心，不在订单中心

   商品的结构远不只是标题和价格，往往还包括：

   • 多规格与规格组合
   • 扩展属性
   • 类目树与类目口径
   • 图文与多媒体资产
   • 履约、交易、营销可见字段

   这些结构的业务语义都属于商品域。如果让订单中心自己去“抓商品再拍快照”，商品领域代码就会反向泄露到交易域，最终形成 DTO 强耦合和边界污染。

2. 交易链路应该传引用，不应该现场组装复杂商品对象

   在高并发下单、秒杀或大促场景里，如果订单中心每次都要实时请求商品中心、拉取完整商品对象、再本地拼成订单商品快照，会把商品中心的读性能直接暴露给交易洪峰。

   相反，如果商品中心在发布时就把 item_snapshot 预先固化好，那么订单中心创单时只需要拿到：

   • item_id
   • item_snapshot_id
   • 当时有效的价格、库存、履约上下文引用

   这样交易链路会退化成结构化标量与引用传递，吞吐量和稳定性都会更好。

3. 订单解释事实必须依赖已冻结的商品快照

   订单一旦成立，之后无论商品标题、图文、规格文案甚至履约说明怎样变化，订单都必须能回到“下单时平台承诺给用户的那一版商品事实”。

   这个“解释事实”应该是商品中心提前冻结好的 item_snapshot，而不是订单中心事后临时拼出来的一份影子 DTO。

设计收束

因此，这个决策点的最终边界可以这样钉死：

• 商品中心负责生成和维护 item_snapshot
• 订单中心只在创单时绑定 item_snapshot_id
• 订单解释基于快照，不基于“当前线上商品”

一句话说：

商品中心负责定义“当时卖的是什么”，订单中心负责记录“用户买的是哪一版”。
快照应该由商品主域提前固化，而不是由交易域在下单瞬间临时拼装。

2.11 决策点 9：供给、商品、库存、搜索、营销、订单的发布一致性应该如何设计

场景与问题

当商家或运营在 B 端点击“发布”时，真正被改变的往往不只是商品中心一张表，而是整条交易主链路上的多个领域：

• 供给平台要推进任务和发布状态
• 商品中心要把 Draft merge 成正式 Item
• 库存中心可能要更新库存配置或资源路由
• 搜索域要刷新索引与检索快照
• 营销域要刷新价格、券适用性或促销上下文
• 订单域要基于最新版本判断确认页快照是否失效

这里最大的风险不是“某一个服务没更新”，而是 多个服务更新节奏不同。

最典型的资损场景是：

• 商品价格已经上调
• 营销优惠先发布生效
• 搜索或商品详情还停留在旧价格
• 用户用旧价叠加大额券瞬间下单

因此，这个决策点要回答的不是抽象的“是否一致”，而是：

供给、商品、库存、搜索、营销、订单等多域之间，应该采用什么样的发布一致性模型，既保证 B 端发布体验，又死守 C 端资金安全。

方案对比

为什么不能用运行时强一致

很多同学在面对“商品、库存、搜索、营销一起发布”时，第一反应是：

用分布式事务把这些服务串起来。

但在真实生产环境里，这几乎一定会出问题。

原因有 3 个：

1. B 端发布体验会被拖死
   • 发布动作会同步等待商品、库存、搜索、营销全部返回
   • 只要其中一个服务网络抖动或处理变慢，整个运营后台就会卡死
2. 跨域长事务会把数据库行锁拖进交易高峰
   • 库存、营销、商品等库都可能被长事务锁住
   • C 端下单、查价、扣库存会被 B 端发布操作反向拖垮
3. 失败点越多，回滚越脆弱
   • 搜索是 ES
   • 营销有缓存和规则引擎
   • 订单有确认页快照
   • 这些本来就不是天然适合 XA 的事务参与者

因此，这里的行业共识应该被明确钉死：

放弃运行时强一致，采用“本地事务 + Outbox 事件广播 + 下游幂等消费 + 版本乱序拦截 + 延迟对账兜底”的发布模型。

推荐方案：基于逻辑版本号驱动的最终一致发布

整套工业级发布一致性链路可以压缩成 4 个动作：

1. 源头状态机推进
   • 供给平台或商品中心只在本地事务内推进自己的发布状态
   • 同时写入 Outbox 事件
2. 事务消息广播
   • 通过 MQ / Outbox 扫描器把商品变更事件 fan-out 给库存、搜索、营销等域
3. 消费端版本校验
   • 下游各域只接受比自己当前 last_processed_version 更新的消息
   • 旧消息、乱序消息直接丢弃
4. 旁路延迟对账
   • 基于 Binlog / Canal / Debezium 或定时审计任务，对多域最终状态做迟到核对与冲正

核心设计点 1：源头必须先状态机化，再发事件

发布动作不能直接“改完数据库顺手发一条 MQ”，而必须先在源头服务中把发布建模成状态机，例如：

• EDITING
• PUBLISHING
• PUBLISHED
• FAILED

以供给平台或商品中心为源头时，本地事务至少做两件事：

1. 推进本地状态到 PUBLISHING
2. 在同一个事务里写一条 ItemChangedEvent 到 Outbox

这一步的关键是：

• 先把本地真相拉平
• 再把“应该通知外界的事实”可靠广播出去

而不是反过来先发消息再祈祷本地和下游都成功。

核心设计点 2：多域广播靠事件，不靠同步 RPC 链式调用

一条商品变更事件发出去后，库存、搜索、营销、数据平台等域应该独立异步消费，而不应该挂成一条“商品调库存、库存调搜索、搜索再调营销”的同步调用链。

这样做的原因是：

• fan-out 广播更容易横向扩展
• 每个域都可以按自己的节奏消费
• 某一个域短时失败不会拖垮整个发布入口

这也意味着：

• 商品中心负责商品正式真相
• 库存域负责库存配置或库存路由真相
• 搜索域负责检索投影
• 营销域负责促销上下文

它们彼此独立，但都围绕同一个 发布版本号 或 逻辑版本号 收敛。

核心设计点 3：下游必须用 version 做乱序拦截

这是整个发布一致性设计里最关键的一道防线。

运营可能在 1 秒内连续改两次价格：

• 第一次把价格改成 100
• 第二次立刻改成 150

如果“150”的消息先到、而“100”的旧消息后到，下游如果无脑覆盖，就会把正确价格覆盖回旧值，直接造成资损。

因此，源头发出的事件必须至少携带：

• item_id / sku_id
• version
• op_time
• 本次变化的核心字段

典型事件体可以抽象成：

{
  "item_id": "ITEM_8888",
  "event_type": "ITEM_UPDATE",
  "price": 15000,
  "version": 1002,
  "op_time": 1780900000
}

下游服务必须在自己的余额表、配置表或投影表里冗余：

• last_processed_version

并在更新时做原子谓词检查：

UPDATE projection_xxx
SET item_price = :price,
    last_processed_version = :version
WHERE item_id = :item_id
  AND :version > last_processed_version;

如果 affected_rows == 0，说明这是一条迟到的老消息，应直接丢弃。

一句话说：

发布一致性的核心不是“消息有没有到”，而是“老消息永远不能覆盖新版本”。

核心设计点 4：订单域必须用确认页快照隔离版本漂移

订单域是最不能“等最终一致慢慢修”的地方，因为用户是在交易链路里直接下单和付钱。

这里不能简单地说“商品最终会一致”，而要明确：

• 用户点击“去结算”时，必须生成一份确认页快照
• 这份快照里锁定当时的 item_id / price / version

后续在 CreateOrder 时，订单服务要拿着这份版本去做最终校验：

• 如果商品仍是同一版本，允许下单
• 如果商品已经推进到更高版本，例如改价或下架，直接提示用户刷新确认页

也就是说：

• 搜索和营销可以接受短暂的投影延迟
• 订单不能接受“旧快照直接落单”

各业务域的协同策略

1. 商品域 → 搜索域

   • 搜索收到变更后批量写 ES
   • 考虑 ES 近实时刷新延迟，可同时刷新 Redis 商品检索快照
   • C 端详情直达优先读 Redis 快照，缓冲 ES 延迟

2. 商品域 → 营销域

   • 营销收到价格、上下架、品类变化后刷新自身投影
   • 但营销在最终核价时，仍应回查或读取高性能缓存里的当前商品正式价格版本
   • 一旦发现版本不一致，应宁可降级不用优惠，也不要放行“低价高券”

3. 商品域 → 库存域

   • 如果商品改动涉及库存配置、履约方式、资源路由或扣减时机，库存域消费事件后应更新自己的配置投影
   • 同样要使用 last_processed_version 防止旧配置覆盖新配置

4. 商品域 → 订单域

   • 订单域不实时追商品所有细节
   • 只在确认页和创单时拿着 version 做关键校验
   • 商品快照与版本对不上时，直接拒绝旧确认页继续下单

终极兜底：旁路延迟对账与自动冲正

即使前面的 Outbox、MQ、版本拦截都做对了，真实生产环境里依然可能发生：

• 消息漏投
• 消费者卡死
• 搜索 / 营销某个域长时间没消费成功

所以还必须有一条不依赖主业务链路的旁路审计机制。

推荐做法是：

1. 在商品发布成功后，基于 Binlog 订阅（Canal / Debezium）或审计任务记录发布事实
2. 延迟几分钟再触发跨域核对，给异步消费预留静默窗口
3. 对库存、搜索、营销、商品详情投影、价格快照等关键域做版本和关键字段核对
4. 一旦发现某个域还停留在旧版本：
   • 触发自动冲正
   • 或调用受控后台接口强制回刷
   • 同时报警

这条链路的定位非常重要：

• 它不是主发布路径
• 它是最终一致性的独立审计者

推荐方案

推荐采用：

本地事务状态机 + Outbox 广播 + 下游版本拦截 + 订单快照隔离 + 延迟对账兜底

这套方案的核心判断是：

1. 发布链路的第一目标不是“所有域瞬时同时成功”，而是 源头状态确定、事件可靠发出
2. 下游一致性的第一目标不是“每条消息都严格按网络顺序到达”，而是 旧版本永远不能覆盖新版本
3. 交易安全的第一目标不是“营销和搜索马上变对”，而是 订单不能基于过期价格版本落单
4. 最终一致性的最后一层，不是业务代码里的 if/else，而是 旁路审计和自动冲正

一句话说：

在超大规模电商里，多域发布一致性不是靠运行时强一致来硬扛，而是靠“版本化事件广播 + 消费端乱序拦截 + 订单快照隔离 + 旁路延迟对账”这套组合拳，把发布体验、系统吞吐和资金安全同时守住。

2.12 状态机与任务模型设计

2.12.1 生命周期不能只用一个大状态字段表达

整套系统里最容易做错的一件事，就是试图用一个大状态字段表达所有生命周期。

但实际上，下面这些状态并不处于同一层：

• DRAFT / QC_PENDING / PUBLISHED
• ONLINE / OFFLINE / BANNED / ARCHIVED
• AVAILABLE / RESERVED / CONFIRMED / RELEASED / CONSUMED
• CREATED / RUNNING / PARTIAL_SUCCESS / FAILED / DONE

如果把它们强行塞进一个统一状态字段里，短期看起来省表、省字段，长期一定会出现：

• 状态爆炸
• 语义互相污染
• 不同系统职责无法切开
• 补偿、审计和恢复策略无从落地

因此这里必须先统一一条系统级原则：

生命周期不是一条线，而是多层状态机的组合。
任务状态、审核状态、正式商品状态、库存状态必须显式分层，不能混成一个大状态字段。

2.12.2 决策点 10：是否只用一个大状态字段表达全部生命周期

推荐方案是方案 B。

原因不在于“拆得越多越优雅”，而在于不同状态层在回答不同问题：

• 任务状态：这次任务有没有执行完
• 审核状态：这次变更有没有被裁判通过
• 正式商品状态：这个商品现在能不能被平台售卖
• 库存状态：这份库存或资源当前处在什么履约阶段

如果把这些问题混在一起，最终就会得到一套谁也解释不清、谁也不敢改的状态系统。

2.12.3 任务状态、审核状态、正式商品状态、库存状态的分层设计

建议至少拆成 4 层：

进一步讲，它们之间的关系应该是：

• 任务状态 驱动“有没有处理完”
• 审核状态 驱动“能不能进入正式发布”
• 正式商品状态 驱动“平台货架能不能卖”
• 库存状态 驱动“用户当前能不能占、能不能扣、能不能核销”

它们彼此关联，但不互相替代。

例如：

• 一个任务可以 FAILED，但商品正式状态仍然是 ONLINE
• 一个商品可以 OFFLINE，但历史库存资源仍可能处于 CONFIRMED 或 CONSUMED
• 一个审核可以 REJECTED，但这并不等于库存要发生任何状态变化

所以这部分最终要落成的系统心智是：

不同状态层服务于不同的治理目标。
任务解决执行闭环，审核解决裁判闭环，商品状态解决售卖闭环，库存状态解决履约闭环。

3. 详细设计 - 供给平台

这一节不再讨论“商品资产最终长什么样”，而是专门回答供给平台如何承接多入口流量、如何隔离不同入口的资源和执行策略、如何把脏输入治理成标准对象，并如何把结果交给商品主域写侧与库存系统。

这里先钉死三个边界：

1. 供给平台负责接入、标准化、任务编排、发布编排和失败运营化。
2. 供给平台不持有 Draft / Staging / QC 主权，这些状态与快照属于商品主域写侧。
3. 供给平台承接库存运营入口，但不持有库存事实、预占、账本和券码池主权。

3.1 平台定位与职责边界

供给平台的定位更接近一个 B 端接入与治理控制面，而不是商品资产或库存事实的权威持有者。

一句话总结：

供给平台负责把外部和 B 端的复杂输入组织成可治理、可编排、可恢复的标准化变更；商品主域写侧负责把这些变更冻结、审核、发布为正式商品；库存系统负责把库存命令落成事实。

3.2 决策点

这一节后面会进入大量实现细节。为了避免决策点散落在各个小节里，供给平台相关的关键取舍统一先收口在 3.2。后续如果继续新增决策点，也优先继续挂到这里。

3.2.1 决策点 1：任务抢占使用 DB 还是 Redis 分布式锁

Parser Worker 抢占 PENDING 任务时，行业里最主流的两套方案就是：

• 基于 MySQL 行锁 / 轻量 CAS 的 DB 抢占
• 基于 Redis SETNX / Redisson 的分布式锁抢占

这两套方案都能做，但它们解决问题的侧重点完全不同。

推荐结论是：

对 Excel 批量导入这类 长生命周期、低 TPS、强一致、强审计 的 B 端任务，优先使用 DB CAS 抢占；
对秒杀、轻量定时任务、超高频短平快加锁场景，才优先考虑 Redis 分布式锁。

原因不在于 Redis 不够快，而在于这类批处理任务真正敏感的不是“抢占吞吐”，而是：

• 状态是否原子闭环
• 容灾恢复是否确定
• 排障路径是否单一
• 是否会因为锁和状态分离而产生脑裂

因此在当前这条链路里，更稳的选择是：

• 任务状态、租约、进度全部沉淀在 product_supply_task
• 通过一条带条件的 UPDATE ... WHERE status='PENDING' 或 lease_until < NOW() 完成抢占
• 由旁路恢复逻辑接管过期租约任务

一句话讲清楚就是：

这条链路要优先追求 状态强一致和容灾确定性，而不是为了追求更高的锁吞吐，把抢占权拆到 Redis 去制造新的双写不一致面。

3.2.2 决策点 2：多入口是否需要按交互频率与数据吞吐量做物理隔离

供给平台的入口很多，但真正困难的地方从来不是“入口多”，而是不同入口的时效预期、吞吐规模、失败容忍度和用户心理完全不同。如果把单品同步提交、Excel 导入、ERP Push、Supplier Pull 全塞进一套线程池、一套队列、一套 Worker，短期看架构很简单，长期一定会在高峰窗口自相残杀。

这里要回答的不是抽象地“要不要隔离”，而是一个更具体的工程问题：

供给平台应该“一套执行池跑所有入口”，还是应该按交互频率和数据吞吐量做物理隔离。

先给结论：

应该按 交互频率 × 数据吞吐量 做物理隔离。
多入口可以复用治理模型，但不应该共用执行资源池、队列和容错策略。

为什么这件事必须作为独立决策点先钉死：

1. 用户心智不同

   单品创建/编辑是秒级心智；Excel 导入是分钟级、盯着进度条的交互心智；ERP Push 只要求快速 ACK 和高可用；Supplier Pull 更偏向长任务、可恢复优先。把这些链路混在一起，本质上是在让最脆弱的交互型链路去为最大吞吐的系统型流量垫背。

2. 资源模型不同

   单品提交主要吃 Web 线程和 RPC RT；Excel 导入吃文件 IO、流式解析和批量落盘；ERP Push 吃 MQ 堆积和 Consumer 匀速消费；Supplier Pull 吃 Batch Worker、Checkpoint、Lease、Snapshot 和外部限流。它们根本不是同一种执行问题。

3. 容错策略不同

   Excel 需要部分成功、错题本和行级修复；ERP/API 需要在签名失败、模板错配、批次结构畸形时整批阻断；Supplier Pull 还要区分批次级、页面级、对象级失败。执行层不隔离，最后连错误模型都会互相污染。

4. 高峰窗口最容易出现跨链路拖垮

   凌晨供应商同步百万级对象、ERP 疯狂重试、Supplier Pull 长时间占用 Worker 时，如果 Excel 导入和单品创建还共用这套执行资源，前台商家可能连一个 20 行的小 Excel 都要排很久，甚至单品保存都会被拖慢。

落地结构：四象限入口设计

整个接入层不应该只按“来源”分类，更应该按 交互频率 × 数据吞吐量 做资源隔离。

落地结构：资源池 / 队列 / Worker 对应关系

隔离不是概念隔离，而是要落到具体执行介质上：

这里要钉死两个底线：

1. Excel 导入与 Supplier ERP/API 推送不共用线程池、队列和容错模型。
2. Supplier Push 和 Supplier Pull 也不应共用同一执行池，因为一个偏实时受理，一个偏长任务恢复。

推荐方案

推荐的落地原则是：

• 治理层复用
  • 统一 task / task_item / publish_record / operation_log
  • 统一校验结果、Diff、发布编排和错误运营化
• 执行层隔离
  • Local 单品：独立同步链路
  • Local Excel：独立 Excel Worker Pool
  • Supplier Push：独立 MQ + Consumer
  • Supplier Pull：独立 Batch Worker + Checkpoint / Lease

3.2.3 决策点 3：Excel 与 ERP/API 是否可以共用同一套容错机制

这是供给平台里最容易被低估的一个设计点。两者都叫“批量数据”，但它们面对的是两种完全不同的用户心智和恢复语义：

• Excel 导入是 交互型导入
• ERP/API 推送是 系统型同步

真正的问题不是“是不是都可能失败”，而是：

供给平台应该给所有批量入口一套统一错误模型，还是应该按交互型导入和系统型同步拆分容错机制。

先给结论：

不应该共用同一套容错机制。
Excel 导入应以部分成功、行级失败、错题本、人工修复为核心；
ERP/API 推送应以关键错误整批阻断、标准错误码、幂等重试、批次隔离为核心。

为什么必须拆：

1. 用户心智不同

   Excel 导入的操作者坐在页面前等待结果，接受“99 行成功、1 行失败，给我错题本”；ERP/API 推送没有人在等进度条，更在意接口高可用、标准错误码和是否能稳定重试。

2. 错误粒度不同

   Excel 更适合行级错误隔离；ERP/API 更适合批次级阻断。把两者混成一种模型，要么 Excel 体验过于僵硬，要么 ERP 批次语义被稀释。

3. 恢复方式不同

   Excel 的恢复动作往往是“下载错题本、修复后重传”；ERP/API 的恢复动作通常是“幂等重试、重新推送整批、按错误码修复上游数据”。

4. 结果输出不同

   Excel 需要错误文件、错题本、行级提示；ERP/API 需要标准响应码、批次状态、重试与死信策略。结果载体本身就不是一回事。

推荐方案

推荐把两类入口的错误模型明确拆开：

• Excel 导入
  • 支持部分成功
  • 允许行级失败继续向下处理
  • 生成错题本和错误文件
  • 支持运营人工修复后再提交
• ERP/API 推送
  • 关键错误整批阻断
  • 返回标准错误码
  • 依赖幂等重试和批次隔离
  • 超过阈值进入死信或问题单

错误分级规则

第 3 节应该明确区分：

• 可行级跳过的错误
  • 字段格式错
  • 个别映射缺失
  • 单行类目异常
• 必须整批失败的错误
  • 签名失败
  • 密钥过期
  • 核心模板不匹配
  • 批次结构畸形

3.2.4 决策点 4：Excel 批量链路为什么不能继续用传统串行流程

很多团队在批量导入场景的第一版实现里，都会自然地选择“一把梭”串行流程：

• 上传文件
• 接口线程直接开始解析
• 逐行调用下游
• 最后在接口或单机后台线程里给出结果

这种做法在数据量小、链路短的时候能跑通，但一旦进入真正的供给平台场景，问题会迅速暴露出来。Excel 批量创建和更新并不是“把单品流程循环 N 次”那么简单，它面对的是：

• 文件 IO 和业务处理耦合
• 大文件导致的内存和线程占用
• 单行失败拖垮整批
• 系统重启后任务蒸发
• 下游抖动时无法削峰和背压
• 结果文件、错题本、部分成功难以可靠生成

所以这里真正要回答的问题不是“要不要异步”，而是：

Excel 批量链路应该继续沿用传统串行处理，还是升级为任务化、分阶段、可恢复的异步流水线。

推荐把 Excel 批量链路升级成：

• 提交任务
• 解析 Excel
• 处理任务
• 产生结果文件

这不是为了“架构更复杂”，而是为了把批量链路从单机脚本思维升级成：

任务化、分阶段、可恢复、可审计、可部分成功 的工业级流水线。

3.2.5 决策点 5：供应商定时同步链路为什么要单独拆分出来

供应商定时同步链路通常来自三方 ERP 的定时批量同步、货期库存批量回传、接口增量推送等场景。它在后端履约阶段，确实可以复用 Excel 批量链路的部分底座，比如：

• 行级 task_item 落盘
• MQ 行级解耦消费
• Diff / 字段主导权 / base_publish_version 校验
• 结果文件或问题单回填

但它在接入层、吞吐模型、容错语义和监控容灾心智上，必须单独拆出来。原因不是“供应商同步更高级”，而是它和 Excel 批量链路面对的是两种完全不同的世界。

先给结论：

供应商定时同步链路应该独立成一条接入链路和执行链路，但可以复用批量治理底座。
也就是“接入层分流，履约层收口”。

为什么要单独拆分，核心是四组对立面：

1. 流量特征不同

   • Excel 是商家手动上传，流量散落、低频、文件大对象。
   • 供应商同步是 ERP 定时洪峰，流量集中、周期性强、瞬时并发高。
   • 如果共用接入池，ERP 洪峰会把 Excel 导入和单品保存一起拖慢。

2. 数据载体不同

   • Excel 的输入是文件流，需要 Parser Worker、流式解析、Checkpoint 和断点续跑。
   • 供应商同步的输入通常是 JSON/RPC 报文或消息，不需要文件解析器。
   • 如果混用，接入层会被迫同时支持文件解析和报文拆分，代码会快速面条化。

3. 容错语义不同

   • Excel 面向人，追求错题本、部分成功、行级修复。
   • 供应商同步面向机器，追求标准错误码、批次阻断、幂等重试。
   • 把两者的错误模型揉成一套，会让人和机器都不好用。

4. 优先级与背压不同

   • Excel 导表通常是高优先级交互任务，要求尽快反馈。
   • 供应商同步更像低优先级高吞吐的后台洪峰，需要更严格的背压和限流。
   • 如果不拆，凌晨同步会抢占商家白天操作的宝贵资源。

所以，供应商同步链路在架构上应该做成：

• 接入层独立
  • Excel 走文件流接入
  • ERP/ISV 走 API / MQ 接入
• 执行层可复用
  • 最终都沉淀为 task_item
  • 后续共用校验、Diff、发布治理和补偿框架
• 监控容灾独立
  • Excel 盯错题本、部分成功和任务进度
  • 供应商同步盯批次水位、checkpoint、死信和重试

一句话收束就是：

Excel 批量链路和供应商定时同步链路可以共用“履约底座”，但不能共用“接入心智”和“容错模型”。
供应商同步必须单独拆分出来，才能保证 ERP 洪峰不会拖垮运营导表体验。

3.2.6 决策点 6：一次同步任务需要几个小时，是否需要任务分片

当一次供应商同步任务可能持续几个小时，甚至覆盖 100w 级资源时，真正要回答的问题不是“能不能把任务切得更碎”，而是：

面对慢源、强频控和长时间排队，最稳的长任务模型应该怎么选。

对某酒店供应商这类慢源供应商，推荐答案不是“任务分片并发拉取”，而是：

• 上游接入层保持 单 Worker 串行 Pull
• 每次只拉一页或一个 cursor
• 每页成功后立即写快照、写状态、丢 MQ
• 下游 Consumer 集群再慢慢做 Diff / publish_version / 商品 RPC upsert

这里的核心不是分片，而是下面 5 个关键点：

1. 统一时刻单例运行与断点续传
   同步窗口可能横跨数小时，昨天的批次没跑完，今天的调度又来了。
   所以 supplier_sync_batch 必须具备 status=RUNNING + lease_until 的租约机制：

   • 如果存在未过期的运行中批次，本次调度直接阻断
   • 如果租约已过期，新 Worker 通过 CAS 接管批次，并从 current_checkpoint 继续跑

2. 非对称双库物化分流
   100w 酒店的巨型异构 JSON 不适合直接压到 MySQL。

   • HBase / Object Store 负责吞 RAW / NORMALIZED 快照
   • MySQL 负责批次、状态、映射和治理锚点 这不是为了“上大数据组件”，而是为了把“存大对象”和“管状态”拆开。

3. 接入层页级物化收拢
   串行拉回一页后，不能做 500 次单条 Upsert。
   更稳的做法是：

   • HBase 侧按页批量刷写原始快照
   • MySQL 侧按页批量 Upsert 状态台账 / 行级锚点
   • 再按页批量投递 MQ 接入层要尽量消灭单条循环写。

4. 时间差下的版本保护
   这类同步里，上游拉取和下游消费之间可能相隔数小时。

   • 供给侧负责忠实搬运，不承担最终过滤主权
   • Consumer 在写商品主域前必须实时反查版本和锁定状态
   • 商品中心在最终写入前用版本、主导权和幂等能力完成最后裁决 目标不是在供给侧裁掉数据，而是防止旧消息在长时间排队后覆盖新版本。

5. 动态多阶计数器与终结标记
   串行 Pull 在拉到最后一页前并不知道总对象数，所以不能沿用“固定总数计数器”的套路。
   更适合的是：

   • Worker 每投一页就 INCRBY
   • 拉到尾页时写入 end_of_stream=true
   • 下游 Consumer 每处理一条就 DECR
   • 计数归零且终结标记为真时，再触发收尾

推荐的长任务结构是：

• 定时任务先创建或接管 supplier_sync_batch
• 一个独占 Worker 串行拉取供应商页面
• 每页成功后立即做页级物化：快照、状态台账、MQ 投递
• current_checkpoint 只记录“已经安全拉回并送入 MQ 的最高水位线”
• RAW / NORMALIZED 快照落 HBase / 对象存储
• MySQL 只保留批次、状态台账、映射、治理锚点
• 下游 Consumer 再慢慢做 Diff / publish_version / 商品 RPC upsert

一句话收束就是：

对于性能一般、频控严格、同步窗口很长的供应商，关键不是任务分片，而是 单例运行、串行拉取、页级物化、下游强幂等与版本拦截。
最合适的结构是“单 Worker 串行 Pull + checkpoint + MQ 异步消化”的线性流水线。

3.2.7 决策点 7：供应商只提供全量 List，如何识别应该下线的对象

供应商同步还有一个非常难啃的问题：很多供应商只会告诉你“我现在还有什么”，不会告诉你“我刚刚删除了什么”。

这会形成典型的“沉默的消失”：

• 创建 / 更新很好处理：List 里有的对象，系统里没有就创建，有了就更新
• 下线最难处理：如果供应商在自己的系统里下掉了一个对象，它不会发 DELETE 事件，而只是让这个对象从下一次 List 里消失

这里真正的难点不是“怎么比一次”，而是：

在供应商只给全量 List 的前提下，如何把“缺失对象识别”设计成可恢复、可审计、可熔断的批次收尾动作。

推荐先给结论：

• 默认采用 时间戳游标对账法
• 对误下线极其敏感的场景，再升级到 影子快照表对账法

为什么默认推荐时间戳游标法？因为它不要求每次都建立完整影子表，也不需要做大范围 NOT IN 式对账。更稳的思路是：

1. 批次启动时记录 T_start
2. 本次 List 中出现并成功进入处理链路的对象，在状态台账里完成“存活打卡”
3. 当且仅当本批次已经 拉取结束 + 下游消费完成收敛，再统一识别那些“本次没有打卡”的在线对象
4. 这些对象进入“候选下线集”
5. 先做比例熔断，再决定是否批量下线

这里有两条必须锁死的安全红线：

1. 绝不能在消费尚未完成时提前下线
   必须等 end_of_stream=true 且对象级处理已收敛，才能开始识别候选下线对象。否则会把还在 MQ 排队的正常对象误杀。

2. 必须有数量安全阈值熔断
   如果本次候选下线对象比例异常高，比如一次要下线 80% 甚至 95%，默认应该认为是供应商漏传、文件不完整或接口故障，而不是“真的大规模下架”。此时必须阻断自动下线，转入告警和人工介入。

所以，这个问题的最终答案不是“做一条大 SQL 找差集”，而是：

用 批次时间锚点 + 对象打卡 + 收尾时统一盘点 + 下线比例熔断，把“沉默的消失”收敛成一个可控、可补偿、可审计的批次收尾步骤。

3.3 统一任务模型与执行框架

入口隔离和执行资源划分已经在 3.2.2 定义完成。这一节开始只讨论统一任务锚点、状态机和抢占模型，而不再重复解释为什么要做四象限隔离。

这里的“统一任务模型”不是指所有入口都共用一套执行线程池，而是指 Local 单品、Excel、API/ISV 推送、运营批量编辑 这些供给治理型入口共用一套治理任务锚点：

• product_supply_task
• product_supply_task_item
• receipt_id
• trigger_id
• operation_id
• execution_mode

共用范围

因此需要在正文中明确一句：

Local 任务与 Supplier 任务在治理层复用发布编排锚点，但执行层不共用线程池、队列和 Worker 模型。

任务状态机

统一任务模型除了统一表结构，还要统一状态机。否则单品、批量、API 推送虽然都写进了 product_supply_task，但每条链路各自发明一套状态语义，最后还是会退化成多套排障和审计口径。

推荐把状态拆成两层：

• product_supply_task 负责表达“这一批整体走到哪里了”
• product_supply_task_item 负责表达“这一行 / 这一对象具体成没成功”

stateDiagram-v2
    [*] --> CREATED
    CREATED --> RUNNING: sync execution starts / worker accepted
    CREATED --> CANCELED: withdrawn before execution
    RUNNING --> DONE: all items succeeded
    RUNNING --> PARTIAL_SUCCESS: some items failed
    RUNNING --> FAILED: batch-level failure or all items failed
    RUNNING --> CANCELED: manually terminated
    PARTIAL_SUCCESS --> DONE: retry succeeded
    PARTIAL_SUCCESS --> FAILED: retry exhausted / manual close
    FAILED --> RUNNING: retry or compensation retry

stateDiagram-v2
    [*] --> PENDING
    PENDING --> VALIDATING
    VALIDATING --> FAILED: validation error
    VALIDATING --> READY: normalized and routable
    READY --> PUBLISHED: publish succeeded
    READY --> FAILED: rejected / publish failed
    READY --> SKIPPED: no-op / ownership deny
    FAILED --> PENDING: item retry

推荐语义如下：

这套状态机要强调两个边界：

• task 状态不等于商品状态。DONE 只表示供给任务处理完了，不代表商品一定已经 ONLINE。
• task_item.READY 也不等于正式发布成功，它只表示该对象已经完成供给平台治理，可以进入商品主域写侧的 Draft / Staging / QC 链路。

3.3.1 任务抢占设计

异步任务模型一旦进入多 Worker 部署，就不能默认“只有一个执行器会来处理这条任务”。如果没有显式抢占机制，最常见的后果就是：

• 两个 Worker 同时处理同一条任务
• 旧进程恢复后覆盖新进程进度
• 任务卡在 RUNNING / PARSING 无法恢复
• 解析进度和执行进度被重复推进

因此，product_supply_task 建议内建一套 数据库权威的任务抢占模型，至少包括：

• worker_id
• lease_token
• lease_until
• heartbeat_at
• parse_checkpoint 或等价进度字段

推荐原则是：

• 任务事实以 MySQL 为权威
• 抢占通过数据库 CAS 完成
• 只有租约持有者才能续租、推进 checkpoint、结束任务
• 只允许抢占租约过期任务，不强抢心跳正常任务

这套设计最好拆成“首次抢占、周期续租、租约过期接管、旧进程失效”四个阶段来看，避免把它误解成一次简单的 UPDATE ... WHERE status='PENDING'。

sequenceDiagram
    autonumber
    participant SCH as "调度器 / 轮询器"
    participant W1 as "Parser Worker-A"
    participant T as "product_supply_task"

    SCH->>T: 创建 task(status=PENDING, parse_checkpoint=0)
    SCH-->>W1: 下发待抢占 task_id
    W1->>T: CAS 抢占任务<br/>PENDING -> PARSING<br/>写 worker_id / lease_token / lease_until / heartbeat_at
    alt rows_affected = 1
        T-->>W1: 抢占成功
        W1->>T: 写首个 parse_checkpoint
        loop 解析期间周期续租
            W1->>T: 更新 heartbeat_at / lease_until / parse_checkpoint
        end
        W1->>T: 解析完成后切换为 PROCESSING
    else rows_affected = 0
        T-->>W1: 抢占失败，任务已被其他 Worker 持有
    end

这一小节最重要的不是“如何生成一个 worker_id”，而是把所有权边界钉死：

这张图只保留任务抢占的最小闭环，重点是：

• 第一次抢占必须由数据库 CAS 原子完成，不能先查再改。
• 续租必须同时更新心跳、租约和 checkpoint。
• 旧 Worker 失去租约后就不能再写回状态。

3.3.2 供给治理任务数据库权威抢占模型

根据 3.2.1 决策点 1：任务抢占使用 DB 还是 Redis 分布式锁，供给平台统一采用 MySQL CAS + Lease 作为任务抢占模型，不再引入 Redis 分布式锁。原因很直接：对于 Excel 批量、运营批量编辑这类 B 端长任务，最重要的不是极限抢占吞吐，而是状态、租约、进度、恢复都必须沉淀在同一权威主表中。

这个模型依赖两个核心前提：

• 数据库是唯一状态权威
• 双因子所有权校验：worker_id + lease_token

worker_id 用来表达“哪台物理执行器正在持有任务”，lease_token 用来表达“这一次抢占行为的唯一租约实例”。只有两者同时匹配，当前 Worker 才有资格续租、推进 checkpoint、回写状态。这样即使旧 Worker 假死后恢复，也无法越权回写。

flowchart TD
    DB["MySQL 任务权威状态台账"] --> CAS["CAS 抢占<br/>status=PENDING 或 lease_until 已过期"]
    CAS --> WA["Parser Worker A<br/>抢占成功"]
    CAS --> WB["Parser Worker B<br/>rows_affected=0 退避"]
    WA --> HB["周期续租 + 推进 checkpoint"]
    HB --> FAIL["Worker OOM / 假死 / 断电"]
    FAIL --> EXPIRE["lease_until 到期"]
    EXPIRE --> TAKEOVER["新 Worker CAS 接管"]
    TAKEOVER --> BLOCK["旧 Worker 恢复后<br/>token 不匹配，DB 物理拦截回写"]

建议 product_supply_task 至少具备下面这些字段，来支撑数据库权威抢占：

ALTER TABLE `product_supply_task`
ADD COLUMN `status` VARCHAR(32) NOT NULL DEFAULT 'PENDING' COMMENT '任务级状态机',
ADD COLUMN `worker_id` VARCHAR(64) NULL COMMENT '当前持有任务的物理节点标识',
ADD COLUMN `lease_token` VARCHAR(64) NULL COMMENT '每次抢占生成的唯一租约令牌(UUID)',
ADD COLUMN `lease_until` DATETIME(3) NULL COMMENT '租约失效绝对截止时间',
ADD COLUMN `heartbeat_at` DATETIME(3) NULL COMMENT '最后一次心跳时间',
ADD COLUMN `parse_checkpoint` VARCHAR(512) NULL COMMENT '解析进度锚点 JSON',
ADD COLUMN `last_heartbeat_stage` VARCHAR(64) NULL COMMENT '最后心跳所处阶段(CLAIMED/PARSING/DISPATCHING)',
ADD INDEX `idx_status_lease` (`status`, `lease_until`);

首次抢占与僵尸接管可以统一为一条 CAS SQL：

public Optional<TaskLease> tryClaimTask(Long taskId) {
    String workerId = localHost() + "_" + threadId();
    String leaseToken = UUID.randomUUID().toString();
    Date now = new Date();
    Date leaseUntil = new Date(now.getTime() + 30_000);

    int rows = db.executeUpdate(
        "UPDATE product_supply_task SET " +
        " status = 'PARSING', worker_id = ?, lease_token = ?, " +
        " lease_until = ?, heartbeat_at = ?, last_heartbeat_stage = 'CLAIMED' " +
        "WHERE id = ? AND (" +
        " status = 'PENDING' OR " +
        " (status IN ('PARSING','PROCESSING') AND lease_until < ?)" +
        ")",
        workerId, leaseToken, leaseUntil, now, taskId, now
    );

    if (rows == 1) {
        return Optional.of(new TaskLease(taskId, workerId, leaseToken));
    }
    return Optional.empty();
}

周期续租和 checkpoint 推进必须绑定双因子所有权：

public boolean renewLease(TaskLease lease, String checkpointJson, int parsedCount) {
    Date now = new Date();
    Date leaseUntil = new Date(now.getTime() + 30_000);

    int rows = db.executeUpdate(
        "UPDATE product_supply_task SET " +
        " lease_until = ?, heartbeat_at = ?, parse_checkpoint = ?, parsed_count = ?, " +
        " last_heartbeat_stage = 'PARSING' " +
        "WHERE id = ? AND worker_id = ? AND lease_token = ? AND status = 'PARSING'",
        leaseUntil, now, checkpointJson, parsedCount,
        lease.getTaskId(), lease.getWorkerId(), lease.getLeaseToken()
    );

    return rows == 1;
}

这套模型最重要的收益不是“谁先抢到了任务”，而是把下面四件事锁死在同一条数据库事实线上：

• 抢占主权
• 续租主权
• 进度推进主权
• 旧进程失效主权

3.3.3 全局发布协同与状态机转移矩阵

根据 3.1 平台定位与职责边界，供给平台不持有正式态商品落库主权，也不持有库存账本主权。它真正持有的是 任务编排主权、状态表达主权、结果归档主权。因此，这里的状态机不能只写成“跑起来了 / 失败了”，而要能明确表达不同阶段的物理意义。

product_supply_task 解决的是“这一批整体走到哪里了”，product_supply_task_item 解决的是“这一行具体成没成功、失败在哪一层”。

任务级状态转移矩阵

行级明细状态转移矩阵

这套矩阵的设计重点有两个：

1. 任务级状态和行级状态必须分层表达。否则你只能知道“这批差不多失败了”，却永远回答不了“哪一行失败、是校验失败还是写入超时”。
2. 超时未知必须留在中间态。RpcTimeoutEvent 不能简单写成 FAILED，否则旁路自愈和假失败拨正都会失去抓手。

3.4 核心功能：单个创建和编辑

3.4.1 场景画像

单个创建和单个编辑都属于强交互、低吞吐、秒级反馈的入口，但它们的难点并不相同：

• 单个创建关注“从无到有”的受理、校验、建草稿和返回受理结果
• 单个编辑关注“基于哪个正式版本改”的 Diff、base_publish_version 和字段主导权

两者的共同点是：

• 前端体验是同步的
• 后端仍然要写轻量 task(total_count=1, execution_mode=SYNC)
• 供给平台负责受理、标准化、校验和 RPC 编排
• Draft / Staging / QC 主权仍在商品主域写侧

3.4.2 单个创建完整时序图

sequenceDiagram
    participant U as "运营/商家"
    participant W as "供给平台 Web"
    participant T as "product_supply_task"
    participant I as "product_supply_task_item"
    participant V as "标准化/校验层"
    participant A as "Supply Application"
    participant P as "商品中心 RPC"
    participant R as "product_supply_request_log"
    participant H as "旁路自愈任务"

    U->>W: 提交单个商品创建表单
    Note over W,T: Tx1：请求受理 + 先落盘
    W->>T: 创建 task(total_count=1, execution_mode=SYNC)
    W->>I: 创建 task_item(object_type=PRODUCT, business_fingerprint)
    W->>V: 标准化输入、模板校验、交易契约校验
    alt 校验失败
        V-->>W: 返回字段级错误
        W->>I: 更新 item=FAILED
        W->>T: 更新 task=FAILED
        W-->>U: 同步返回失败原因
    else 校验通过
        Note over V,A: Tx2：组装创建命令
        V->>A: 组装 CreateDraftCommand
        A->>R: 记录请求日志(request_id, payload_hash)
        A->>P: CreateProductDraft RPC
        alt 明确成功
            P-->>A: 返回 draft_id / goods_id(可选) / audit_route
            Note over A,T: Tx3：回填 RPC 结果
            A->>R: 记录 response_code=SUCCESS
            A->>I: 更新 item=SUBMITTED，记录 draft_id
            A->>T: 更新 task=SUBMITTED，记录 draft_id
            A-->>W: 返回受理结果
            W-->>U: 返回 draft_id / task_id
        else 超时未知 / 假失败
            P--xA: timeout / unknown
            A->>R: 记录 response_code=TIMEOUT_UNKNOWN
            A->>I: 更新 item=PROCESSING
            A->>T: 更新 task=PROCESSING
            A-->>W: 返回处理中 / 请稍后刷新
            H->>P: 依据 request_id / business_fingerprint 只读反查
            alt 商品中心已成功受理
                P-->>H: 返回 draft_id
                H->>I: 回填 item=SUBMITTED, draft_id
                H->>T: 回填 task=SUBMITTED
            else 商品中心未成功创建
                P-->>H: not found / failed
                H->>I: 回填 item=FAILED
                H->>T: 回填 task=FAILED
            end
        end
    end

3.4.3 单个编辑完整时序图

sequenceDiagram
    participant U as "运营/商家"
    participant W as "供给平台 Web"
    participant T as "product_supply_task"
    participant I as "product_supply_task_item"
    participant V as "标准化/校验层"
    participant D as "Diff/字段主导权"
    participant A as "Supply Application"
    participant P as "商品中心 RPC"
    participant R as "product_supply_request_log"
    participant H as "旁路自愈任务"

    U->>W: 提交单个商品编辑
    W->>T: 创建 task(total_count=1, execution_mode=SYNC)
    W->>I: 创建 task_item(object_type=PRODUCT, base_publish_version, business_fingerprint)
    W->>V: 标准化输入并做格式/契约校验
    V->>D: 计算 Diff，判断字段主导权和风险级别
    alt 无有效变更
        D-->>W: 返回 no-op
        W->>I: 更新 item=SKIPPED
        W->>T: 更新 task=DONE
        W-->>U: 返回无变更
    else 有效变更
        D->>A: 组装 UpdateDraftCommand(base_publish_version)
        A->>R: 记录请求日志(request_id, base_publish_version)
        A->>P: UpdateProductDraft RPC
        alt 明确成功
            P-->>A: 返回 draft_id / audit_route / version_status
            A->>R: 记录 response_code=SUCCESS
            A->>I: 更新 item=SUBMITTED
            A->>T: 更新 task=SUBMITTED
            A-->>W: 返回受理结果
            W-->>U: 返回 draft_id / task_id
        else 版本冲突
            P-->>A: VERSION_CONFLICT
            A->>R: 记录 response_code=VERSION_CONFLICT
            A->>I: 更新 item=FAILED
            A->>T: 更新 task=FAILED
            A-->>W: 返回版本已变化，请刷新后重试
        else 超时未知 / 假失败
            P--xA: timeout / unknown
            A->>R: 记录 response_code=TIMEOUT_UNKNOWN
            A->>I: 更新 item=PROCESSING
            A->>T: 更新 task=PROCESSING
            A-->>W: 返回处理中 / 请稍后刷新
            H->>P: 依据 request_id / business_fingerprint 只读反查
            alt 商品中心已成功受理且版本未冲突
                P-->>H: 返回 draft_id / current_publish_version
                H->>I: 回填 item=SUBMITTED, draft_id
                H->>T: 回填 task=SUBMITTED
            else 商品中心版本已演进
                P-->>H: VERSION_CONFLICT
                H->>I: 回填 item=FAILED
                H->>T: 回填 task=FAILED
            else 商品中心未成功处理
                P-->>H: not found / failed
                H->>I: 回填 item=FAILED
                H->>T: 回填 task=FAILED
            end
        end
    end

3.4.4 关键设计点

3.4.5 表和版本映射

3.5 核心功能：Excel 批量创建和更新

3.5.1 场景画像

Excel 批量创建和更新属于高交互异步任务：用户上传文件后会盯着进度条等结果，期望分钟级反馈，并且接受“部分成功、部分失败、最后给我错题本”。

这条链路真正的工业级做法，不是“上传文件后一个线程从头跑到尾”，而是拆成四个阶段：

1. 提交任务：只收单，不处理，秒级返回 task_id
2. 解析 Excel：流式解析，行级明细落盘
3. 处理任务：通过 MQ 行级解耦，Consumer 匀速顶下游
4. 产生结果文件：汇总成功/失败/跳过，生成错题本

这条链路的核心不是“批量更大”，而是：

• 任务级状态和行级状态必须分离
• 文件 IO、行级处理、商品中心 RPC、结果归档必须分阶段解耦
• 创建和更新会混在同一批文件里
• 更新行要带上 base_publish_version 或当前正式对象引用
• 处理阶段要靠 MQ 削峰和背压，不能靠单机线程池硬顶
• 收尾阶段要能在分布式消费结束后可靠触发，而不是人工扫表碰运气
• 创建和更新虽然共享同一批量底座，但在阶段三“处理任务”的治理内核必须分流

这里不再重复讨论为什么 Excel 批量链路必须任务化、分阶段和异步化，这个架构取舍已经在 3.2.4 决策点 4：Excel 批量链路为什么不能继续用传统串行流程 中统一说明。

3.5.2 批量任务状态机

Excel 批量任务必须显式区分 任务级状态机 和 行级状态机。否则系统只能告诉运营“这批任务大概失败了”，却回答不了“哪一行失败、失败在哪个阶段、是否还能继续重试”。

推荐把这一节的状态机写成更贴近 Excel 批量场景的两层：

stateDiagram-v2
    [*] --> PENDING
    PENDING --> PARSING: parser worker claimed
    PENDING --> CANCELED: withdrawn before parse
    PARSING --> PROCESSING: rows persisted and MQ dispatched
    PARSING --> FAILED: parse fatal error
    PROCESSING --> GENERATING_RESULT: all items finished and result file ready to aggregate
    PROCESSING --> FAILED: batch-level fatal error or all items failed
    PROCESSING --> CANCELED: manually terminated
    GENERATING_RESULT --> DONE: all items succeeded / no error file required
    GENERATING_RESULT --> PARTIAL_SUCCESS: some items failed or skipped, error file generated
    PARTIAL_SUCCESS --> DONE: manual repair or retry succeeded
    PARTIAL_SUCCESS --> FAILED: retry exhausted / force close
    FAILED --> PROCESSING: compensation retry / replay

stateDiagram-v2
    [*] --> INIT
    INIT --> PROCESSING: consumer accepted
    PROCESSING --> SUBMITTED: draft command accepted
    PROCESSING --> SKIPPED: no-op / ownership deny
    PROCESSING --> FAILED: validation error / publish error
    FAILED --> INIT: row retry

这两层状态分别解决不同问题：

这套状态机有 3 个关键点：

1. PARSING 是 Excel 批量独有的重要中间态，单个同步任务通常不会经历这个阶段。
2. PARTIAL_SUCCESS 是批量任务里最关键的任务级状态之一，单个任务几乎不会真正用到它。
3. task_item 的状态必须能独立重试，否则错题本修复和局部补偿就无从谈起。
4. 创建和更新可以共用这套状态机外壳，但更新链路通常会多一个“实时补齐版本 + Diff 裁剪”的隐含处理层。
5. GENERATING_RESULT 是批量任务专有的收尾态，它把“行级消费完成”和“任务最终归档”拆开，避免把结果文件生成逻辑硬塞回 PROCESSING。

3.5.3 批量创建与批量更新的核心差异

批量创建和批量更新虽然共用同一套四阶段异步流水线外壳，但在阶段三“处理任务”的内核策略完全不同。两者最本质的差别，不是谁调用了不同的 RPC，而是它们对商品主数据做的是两种相反的物理动作：

• 批量创建追求 完备性与防冗余
• 批量更新追求 增量安全与抗并发

也正因为如此，同一套 Excel 跑批外壳下，创建和更新必须走不同的卡点设计。

差异 1：对象存在性的前置判断不同

• 批量创建：
  • 逻辑是“对象绝对不该存在”
  • Consumer 拿到一行后，先基于 business_fingerprint 或等价唯一特征反查
  • 如果系统里已经存在同指纹商品，该行直接失败并进入错题本，防止重复创建
• 批量更新：
  • 逻辑是“对象必须存在”
  • Consumer 拿着 goods_id / item_id / spu_id 等正式对象标识反查
  • 如果根本找不到对应正式商品，该行直接失败，因为失去了更新主体

差异 2：版本号的防御姿态不同

• 批量创建：
  • 不依赖 base_publish_version
  • 属于从 0 到 1 的初始化过程
  • 创建命令只要满足完备性和防重要求即可
• 批量更新：
  • 必须在消费阶段实时反查并补齐 base_publish_version
  • 更新命令必须显式带上版本约束
  • 这是更新链路最关键的抗并发底线，用来阻断旧版本覆盖新版本

差异 3：数据剪裁与组装机制不同

• 批量创建：
  • 更像胖报文
  • 关注“合规的最小完备集”
  • 少一个必填项就应直接失败
• 批量更新：
  • 更像瘦报文
  • 先做 Diff
  • 再做字段主导权裁剪
  • 再做缺失字段补全
  • 绝不能用 Excel 里的空字段覆盖线上已有值

差异 4：并发与锁竞争退避策略不同

• 批量创建：
  • 风险点是重复创建和指纹碰撞
  • 更依赖本地 task_item 唯一指纹索引
  • 同批内还可以先做内存去重
• 批量更新：
  • 风险点是下游已存在热点行的锁竞争
  • 更需要 merchant_id / item_id 局部 Hash 顺序路由
  • 把无序并发写驯化成顺序消费，给下游主库卸压

一句话收束就是：

批量创建和批量更新共享的是同一种跑批外壳，但在处理内核上必须做“非对称卡点”：创建防冗余，更新防并发。

3.5.4 任务抢占与批量处理时序图

3.5.4.1 任务抢占时序图

sequenceDiagram
    autonumber
    participant SCH as "调度器 / 轮询器"
    participant W1 as "Parser Worker-A"
    participant W2 as "Parser Worker-B"
    participant MON as "任务巡检器"
    participant T as "product_supply_task"

    SCH->>T: 创建 task(status=PENDING, parse_checkpoint=0)
    SCH->>T: 周期扫描候选任务<br/>status = 'PENDING' AND task_type = 'IMPORT'
    SCH-->>W1: 下发待抢占 task_id

    W1->>T: CAS 抢占任务<br/>UPDATE ... SET status='PARSING', worker_id, lease_token,<br/>lease_until, heartbeat_at, parse_checkpoint<br/>WHERE id=? AND status='PENDING'
    alt rows_affected = 1
        T-->>W1: 抢占成功，任务所有权归属 W1
        W1->>T: 进入解析前初始化<br/>last_heartbeat_stage='CLAIMED'
        W1->>T: 写首个 parse_checkpoint<br/>sheet=1,row_no=1,byte_offset=0

        loop 解析期间周期续租
            W1->>T: 更新 heartbeat_at / lease_until
            W1->>T: 推进 parse_checkpoint / parsed_count
            T-->>W1: rows_affected = 1
        end

        alt 解析完成
            W1->>T: 提交解析结果<br/>parsed_count / success_count / failed_count
            W1->>T: 切换 task 状态<br/>PARSING -> PROCESSING
            W1->>T: 保留 worker_id / lease_token<br/>供后续处理阶段复用
        else 解析异常 / OOM / 进程退出
            W1--x T: 心跳停止，lease_until 逐渐到期
        end
    else rows_affected = 0
        T-->>W1: 抢占失败，任务已被其他 Worker 持有
        W1-->>SCH: 放弃执行，等待下次调度
    end

    MON->>T: 轮询僵尸任务<br/>status in ('PARSING','RUNNING') AND lease_until < now()
    alt 发现租约已过期
        MON-->>W2: 通知可接管任务
        W2->>T: CAS 抢占过期任务<br/>WHERE id=? AND status IN ('PARSING','RUNNING')<br/>AND lease_until < now()
        alt 接管成功
            T-->>W2: rows_affected = 1
            W2->>T: 覆盖 worker_id / lease_token / lease_until / heartbeat_at
            W2->>T: 从 parse_checkpoint 继续解析<br/>允许局部重跑一小段
        else 任务已被别的 Worker 接管
            T-->>W2: rows_affected = 0
            W2-->>MON: 放弃接管
        end
    else 心跳仍正常
        MON-->>T: 记录任务健康，无需接管
    end

    W1->>T: 旧进程恢复后尝试续租或回写
    T-->>W1: worker_id + lease_token 不匹配，更新失败
    W1-->>SCH: 旧进程自动失效，不可写回状态

3.5.4.2 批量创建完整时序图

sequenceDiagram
    autonumber
    participant U as "运营/商家"
    participant W as "供给平台 Web"
    participant OSS as "OSS / S3"
    participant T as "product_supply_task"
    participant I as "product_supply_task_item"
    participant PX as "Parser Worker"
    participant MQ as "Task Item MQ"
    participant C as "Item Consumer"
    participant V as "标准化/校验层"
    participant P as "商品中心 RPC"
    participant RC as "Redis Counter"
    participant F as "Result File Worker"

    U->>OSS: 上传 Excel
    OSS-->>U: 返回 file_url
    U->>W: 提交 file_url / 发起批量创建
    W->>T: 创建 task(status=PENDING, execution_mode=ASYNC)
    W-->>U: 返回 task_id / receipt_id
    W->>PX: 投递解析任务
    PX->>T: CAS 抢占任务(PENDING -> PARSING, worker_id, lease_token)
    PX->>T: 更新 heartbeat_at / parse_checkpoint
    loop 每一行
        PX->>I: 批量写入 task_item(status=INIT, raw_row, normalized_snapshot)
        PX->>T: 批量更新 parsed_count / parse_checkpoint / heartbeat_at
    end
    PX->>T: 更新 task=PROCESSING, total_count
    PX->>RC: SET task:count:{task_id}=total_count
    PX->>MQ: 投递每个 task_item_id
    loop 每个 task_item 消费
        C->>MQ: 消费 task_item_id
        C->>I: 读取 task_item，更新 item=PROCESSING
        C->>V: 标准化校验 / 模板校验 / 契约校验
        alt 行级校验失败
            V-->>I: 更新 item=FAILED，记录 error_code / error_message
        else 创建行
            V->>V: 基于 business_fingerprint 反查是否已存在
            V->>P: CreateProductDraft RPC
            P-->>I: 回填 draft_id / audit_route，更新 item=SUBMITTED
        else 临时性网络失败
            C-->>MQ: RECONSUME_LATER
        end
        C->>RC: DECR task:count:{task_id}
        alt 计数归零
            RC-->>C: 0
            C->>F: 触发结果归档
        end
    end
    F->>I: 汇总 success / failed / skipped
    F->>OSS: 生成并上传错题本 Excel
    F->>T: 回填 success_count / failed_count / skipped_count / error_file_url / status=DONE
    T-->>U: 前端轮询得到部分成功或全部完成

3.5.4.3 批量编辑完整时序图

sequenceDiagram
    autonumber
    participant U as "运营/商家"
    participant W as "供给平台 Web"
    participant OSS as "OSS / S3"
    participant T as "product_supply_task"
    participant I as "product_supply_task_item"
    participant PX as "Parser Worker"
    participant MQ as "Task Item MQ"
    participant C as "Item Consumer"
    participant V as "标准化/校验层"
    participant D as "Diff/字段主导权"
    participant P as "商品中心 RPC"
    participant RC as "Redis Counter"
    participant F as "Result File Worker"

    U->>OSS: 上传 Excel
    OSS-->>U: 返回 file_url
    U->>W: 提交 file_url / 发起批量编辑
    W->>T: 创建 task(status=PENDING, execution_mode=ASYNC)
    W-->>U: 返回 task_id / receipt_id
    W->>PX: 投递解析任务
    PX->>T: CAS 抢占任务(PENDING -> PARSING, worker_id, lease_token)
    PX->>T: 更新 heartbeat_at / parse_checkpoint
    loop 每一行
        PX->>I: 批量写入 task_item(status=INIT, raw_row, normalized_snapshot)
        PX->>T: 批量更新 parsed_count / parse_checkpoint / heartbeat_at
    end
    PX->>T: 更新 task=PROCESSING, total_count
    PX->>RC: SET task:count:{task_id}=total_count
    PX->>MQ: 投递每个 task_item_id
    loop 每个 task_item 消费
        C->>MQ: 消费 task_item_id
        C->>I: 读取 task_item，更新 item=PROCESSING
        C->>V: 标准化校验 / 模板校验 / 契约校验
        alt 行级校验失败
            V-->>I: 更新 item=FAILED，记录 error_code / error_message
        else 更新行
            V->>P: 反查正式对象 / 当前 publish_version
            V->>D: 补齐 base_publish_version，计算 Diff / 字段主导权
            alt 无有效变更
                D-->>I: 更新 item=SKIPPED
            else 有效变更
                D->>P: UpdateProductDraft RPC
                P-->>I: 回填 draft_id / audit_route，更新 item=SUBMITTED
            end
        else 临时性网络失败
            C-->>MQ: RECONSUME_LATER
        end
        C->>RC: DECR task:count:{task_id}
        alt 计数归零
            RC-->>C: 0
            C->>F: 触发结果归档
        end
    end
    F->>I: 汇总 success / failed / skipped
    F->>OSS: 生成并上传错题本 Excel
    F->>T: 回填 success_count / failed_count / skipped_count / error_file_url / status=DONE
    T-->>U: 前端轮询得到部分成功或全部完成

3.5.5 关键设计点

3.5.5.1 提交阶段的设计点

3.5.5.2 Parser Worker 解析阶段的设计点

3.5.5.3 处理与收尾阶段的设计点

3.5.5.4 结果文件与收尾阶段的设计点

这一阶段的核心目标不是“再处理一批数据”，而是把分布式消费后的行级结果重新收口成一个商家可感知的最终产物。也就是：由谁来判断批次已经全部完成，谁来汇总成功/失败/跳过，谁来生成可下载的错题本 Excel，并把结果回填到主任务表。

推荐做法是把“完成判定”和“结果文件生成”拆成两步：

1. 在解析阶段结束后，由 Parser Worker 先把 Redis 原子计数器初始化为本批次总行数，例如 SET task:count:{task_id} = total_count。
2. 所有 task_item 被 MQ Consumer 处理完后，无论成功、失败还是跳过，都在本地事务回写行级状态，然后执行一次 DECR task:count:{task_id}。
3. 当某个 Consumer 执行 DECR 后拿到的结果正好等于 0，它就成为本批次的“终结者线程”，并先把主任务切换为 GENERATING_RESULT。
4. 终结者线程异步汇总 product_supply_task_item 的行级状态，计算 success_count / failed_count / skipped_count。
5. 如果存在失败行，就按照原始行数据 + error_code / error_message 流式生成错题本 Excel，上传到 OSS/S3，得到 error_file_url。
6. 最后由终结者线程在一个轻量本地事务里回填 product_supply_task：status=DONE 或 status=PARTIAL_SUCCESS、success_count、failed_count、skipped_count、error_file_url。

如果 Redis 计数器因为漏计、超时或者 Consumer 猝死没有自然归零，系统还需要一个旁路兜底定时器定期扫描 PROCESSING / GENERATING_RESULT 的任务和 task_item 状态：

• 如果 DB 聚合结果已经显示所有行都处理完成，但 Redis 计数器没归零，则由定时器补触发一次 GENERATING_RESULT 和结果文件生成。
• 如果 DB 仍然有未完成行，则任务继续保持 PROCESSING，等待后续 Consumer 或恢复 Worker 继续完成。

这也是为什么“Redis 闪电收尾 + DB 旁路盘点”必须同时存在：

• Redis 负责高性能、低成本地找出最后一个完成者。
• DB 负责在异常场景下给出最终真相，避免计数器漏计把结果文件永远卡住。
• 结果文件生成不是一个纯内存事件，而是一个必须能被数据库复核、补触发、补回填的任务收尾动作。

3.5.6 批量链路交付级核心表结构

对于 Excel 批量链路，真正的底座就是两张表：

• product_supply_task：承载任务级状态机、双因子租约、错题本与统计计数
• product_supply_task_item：承载行级明细沙盒、幂等、防冗余、错误归因

下面这组 DDL 不要求你逐字段 1:1 照搬，但字段职责最好不要偏离。

CREATE TABLE `product_supply_task` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `receipt_id` VARCHAR(64) NOT NULL COMMENT '对外受理凭证号',
  `task_type` VARCHAR(32) NOT NULL DEFAULT 'IMPORT' COMMENT 'IMPORT / SYNC / BATCH_EDIT',
  `execution_mode` VARCHAR(32) NOT NULL DEFAULT 'ASYNC' COMMENT 'SYNC / ASYNC',
  `source_type` VARCHAR(32) NOT NULL COMMENT 'MERCHANT_BACKEND / OPERATOR_ADMIN / ERP_API / SUPPLIER_PULL',
  `operator_id` VARCHAR(64) NOT NULL COMMENT '操作人/系统标识',
  `merchant_id` BIGINT NOT NULL COMMENT '商家/供应商ID',
  `status` VARCHAR(32) NOT NULL DEFAULT 'PENDING' COMMENT '任务级状态机',
  `worker_id` VARCHAR(64) DEFAULT NULL COMMENT '当前持有任务的物理节点',
  `lease_token` VARCHAR(64) DEFAULT NULL COMMENT '当前租约令牌',
  `lease_until` DATETIME(3) DEFAULT NULL COMMENT '租约到期时间',
  `heartbeat_at` DATETIME(3) DEFAULT NULL COMMENT '最后心跳时间',
  `last_heartbeat_stage` VARCHAR(64) DEFAULT NULL COMMENT '最后心跳阶段',
  `parse_checkpoint` VARCHAR(512) DEFAULT NULL COMMENT '解析断点 JSON',
  `file_url` VARCHAR(512) DEFAULT NULL COMMENT '原始 Excel 文件路径',
  `error_file_url` VARCHAR(512) DEFAULT NULL COMMENT '错题本路径',
  `total_count` INT NOT NULL DEFAULT 0 COMMENT '总有效行数',
  `parsed_count` INT NOT NULL DEFAULT 0 COMMENT '已解析行数',
  `success_count` INT NOT NULL DEFAULT 0 COMMENT '成功行数',
  `failed_count` INT NOT NULL DEFAULT 0 COMMENT '失败行数',
  `skipped_count` INT NOT NULL DEFAULT 0 COMMENT '跳过行数',
  `gmt_create` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `gmt_modified` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_receipt_id` (`receipt_id`),
  KEY `idx_status_lease` (`status`, `lease_until`),
  KEY `idx_merchant_create` (`merchant_id`, `gmt_create`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供给平台统一任务控制主表';

CREATE TABLE `product_supply_task_item` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '明细主键ID',
  `task_id` BIGINT UNSIGNED NOT NULL COMMENT '关联任务ID',
  `row_number` INT NOT NULL COMMENT 'Excel 绝对行号',
  `object_type` VARCHAR(32) NOT NULL DEFAULT 'PRODUCT' COMMENT 'PRODUCT / COMBINED_PRODUCT / INVENTORY_COMMAND',
  `goods_id` BIGINT UNSIGNED DEFAULT NULL COMMENT '正式商品ID',
  `business_fingerprint` VARCHAR(64) NOT NULL COMMENT '业务指纹',
  `base_publish_version` INT UNSIGNED DEFAULT NULL COMMENT '更新链路版本锚点',
  `status` VARCHAR(32) NOT NULL DEFAULT 'INIT' COMMENT '行级状态机',
  `raw_row_data` LONGTEXT DEFAULT NULL COMMENT '原始行数据快照',
  `normalized_snapshot` LONGTEXT DEFAULT NULL COMMENT '标准化后 DTO 快照',
  `error_code` VARCHAR(64) DEFAULT NULL COMMENT '结构化错误码',
  `error_message` VARCHAR(1024) DEFAULT NULL COMMENT '友好错误提示',
  `gmt_create` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `gmt_modified` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_task_fingerprint` (`task_id`, `business_fingerprint`),
  KEY `idx_task_status` (`task_id`, `status`),
  KEY `idx_goods_id` (`goods_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供给平台任务行级明细沙盒表';

这两张表最重要的设计收益不是“字段很全”，而是：

• 抢占、续租、解析、处理、收尾全都能在数据库里找到权威锚点
• 单行失败、局部重试、错题本回放都不需要回头翻日志
• task 和 task_item 的边界天然支持部分成功和补偿闭环

3.5.7 四阶段流水线的全生命周期战术推演

这一节把前面零散出现的“提交、解析、处理、收尾”收成一条完整流水线，方便直接指导实现。

阶段 1：提交任务（收单解耦面）

• 商家点击上传并提交
• 供给平台 Web 只校验扩展名、文件大小、模板基础合法性
• 创建唯一 receipt_id
• 在 product_supply_task 中插入一条 status=PENDING 的任务
• 立即返回 receipt_id + task_id

这一阶段的目标不是处理业务，而是把“用户提交”可靠地转换成一个可追踪、可抢占、可恢复的任务锚点。

阶段 2：解析 Excel（流式物化面）

• Parser Worker 通过 3.3.2 的 CAS + Lease 模型抢占任务
• 抢占成功后，把任务推进到 PARSING
• 使用流式解析组件读取 Excel，而不是一次性全量读入内存
• 每解析一批行，就批量插入 product_supply_task_item
• 同时推进 parse_checkpoint
• 解析完成后，把 total_count 写入 Redis 计数器，并把任务切到 PROCESSING
• 随后批量投递 task_item_id 到 MQ

这里的关键不是“能读 Excel”，而是把文件 IO 风险安全地转化成数据库里的原子行记录。

阶段 3：处理任务（异步治理分流面）

• Item Consumer 集群根据下游商品主域写侧的承载能力匀速消费
• 创建行走“完备性 + 防冗余”路径
• 更新行走“版本补齐 + Diff + 字段主导权裁剪”路径
• 行处理结束后，立刻回写 task_item.status / error_code / error_message
• 然后执行一次 DECR task:count:{task_id}

这一步最重要的是 行级沙盒隔离：

• 一行失败，只影响这一行
• 不会把整批任务拖回滚
• 也不会阻断其他行继续向前推进

阶段 4：产生结果文件（分布式收尾归档面）

• 某个 Consumer 执行 DECR 后返回值正好等于 0
• 它成为“终结者线程”
• 先把主任务从 PROCESSING 切到 GENERATING_RESULT
• 然后聚合 task_item 状态
• 若存在失败行，则用 raw_row_data + error_message 流式生成错题本 Excel
• 上传 OSS / S3，得到 error_file_url
• 最后回填主任务终态：

UPDATE product_supply_task
SET status = IF(failed_count > 0, 'PARTIAL_SUCCESS', 'DONE'),
    success_count = ?,
    failed_count = ?,
    skipped_count = ?,
    error_file_url = ?
WHERE id = ?;

如果 Redis 计数器漏计、Consumer 猝死或最后一个 DECR 没有自然归零，则由旁路定时器兜底：

• 扫描 PROCESSING / GENERATING_RESULT
• 聚合 task_item
• 如果 DB 已确认没有未完成行，就补触发结果文件生成与主任务回填

因此这一阶段不是“顺手做个文件导出”，而是整条异步流水线从分布式并发重新收敛为可交付结果的总收官。

3.5.8 表和版本映射

3.6 核心功能：供应商同步

3.6.1 场景画像

供应商同步是整个供给平台里最复杂的一类自动化链路。它不只是 “Pull 一批 JSON 回来”，而是同时覆盖：

• Supplier Push
• Supplier Pull
• Full Sync
• Incremental Sync

这条链路为什么要单独拆出来，在 3.2.5 已经做过统一决策：它可以复用批量治理底座，但不能和 Excel 批量共用接入层、线程池和错误模型。这里开始只讲供应商同步链路本身的实现。

其中，某酒店供应商这类慢源 Pull 长任务最有代表性。它和单商品创建、Excel 导入最大的不同，不是“批量更大”，而是同时叠加了这些现实约束：

• 供应商接口性能一般，QPS 严格受限
• 同步窗口可能长达数小时，甚至 10 小时以上
• 外部供应商接口不稳定，可能限流、超时、5xx、cursor 失效
• 上游拉取与下游消费天然存在时间差，旧数据覆盖新数据是核心资损风险
• 任务中断后必须能从 current_checkpoint 恢复，而不是从头重跑
• 供应商原始返回值必须保留为证据和回放素材

为了把这类长任务讲清楚，这里把供应商同步统一抽象成 4 个阶段：

1. Phase 0: Sharder，决定本次同步边界和分片颗粒度。
2. Phase 1: Fetcher，按照分片上下文去外部供应商拉原始数据。
3. Phase 2: Transformer，把原始 JSON / XML 映射成平台标准对象。
4. Phase 3: Publisher，把标准对象持久化、投递、发布到商品主域和治理链路。

这 4 个阶段不是要求所有供应商都做成高并发切片框架，而是提供一个统一的执行心智。对某酒店供应商这类慢源 Pull，Phase 0 仍然可以只产出一个或少量串行分片，Phase 1 保持单 Worker 慢拉，后面的 Transformer / Publisher 再通过 MQ 并发消峰。

这里直接定调：

对某酒店供应商这类慢源，正确答案不是“Fetcher 阶段盲目并发拉”，而是“Sharder 控边界，Fetcher 串行慢拉，Publisher 之后再用 MQ 蓄水和下游并发消峰”。

所以供应商同步这类链路的本质不是“收一批外部数据”，而是：

如何在超长任务、不稳定外部系统和正式发布治理之间，实现 可恢复、可追溯、可治理、可补偿。

3.6.2 整体设计：关键决策点

在进入具体执行模型之前，这一节先不急着铺所有细节，而是先把最重要的判断收口成一组“执行摘要”。这一节主要回答两件事：

• 整体设计上，这条同步链路为什么要拆成四阶段，而不是继续停留在单 Worker 黑盒。
• 关键决策上，为什么治理粒度要做到主任务、批次、分片、对象四层，而不是只做粗粒度状态。

目标只有一个：让读者先看懂这条同步链路为什么要这样分层、为什么治理粒度要做到这里，然后再进入后面的任务治理、分层展开、时序图和表结构。

3.6.2.1 四阶段核心职责边界

这个拆分的最大价值在于：把“拉数据的不稳定”“翻译的复杂性”“发布的治理”彻底解耦。对于某酒店供应商这类慢源、大体量、强治理场景，这种分层方式的长期维护性会明显高于“大 Consumer 泥球”。

后文所有设计点，本质上都是围绕这四个职责边界展开，而不是额外发明新的执行层。

3.6.2.2 总体架构选择：为什么不是纯串行黑盒

在真正展开四阶段执行模型之前，需要先回答一个更上层的问题：

供应商同步到底应该继续采用“一个 Worker 从头拉到尾”的纯串行模式，还是应该升级成“分阶段 + 分片 + 中间状态可观测”的流水线模式。

这是一个非常典型的架构分水岭。推荐先把两个方案明确摆出来：

如果只从“最容易上线”看，方案 A 当然更轻；但一旦同步对象规模进入 100w 级、需要长期稳定运行、需要失败治理和人工介入，方案 A 很快就会暴露两个根本问题：

1. 它只能回答“任务有没有跑完”，很难回答“卡在哪、为什么卡、能不能局部恢复”。
2. 它把拉取、转换、发布和治理揉成一个执行体，后续每加一层能力，复杂度都会继续堆在同一个 Worker 里。

也正因为如此，这一节最终选择方案 B 作为主方案。但这里也要讲清楚一个非常重要的现实判断：

选择方案 B，不等于每个供应商都必须在每个阶段都高并发运行。

对某酒店供应商这种慢源、QPS 严格受限、上游接口不稳定的场景，更合理的落地方式其实是：

• 在架构层选择方案 B，保留分阶段、可治理、可补偿的能力。
• 在执行层让 Fetcher 继续采用偏串行的慢拉模式。
• 把并发能力主要释放到 Transformer / Publisher 和错误处理链路，而不是上游盲目并发拉取。

所以更准确的工程判断应该是：

不是“方案 A 或方案 B 二选一”，而是“整体架构采用方案 B，但对慢源供应商在局部执行策略上保留方案 A 的串行心智”。

这样做的结果是：

• 既保住了某酒店供应商慢源场景下最关键的稳定性和 checkpoint 恢复能力；
• 又不会把整个系统锁死在“单 Worker 黑盒长任务”的演进路径上。

3.6.2.3 任务治理粒度：主任务、批次、分片、对象是否都要建模

如果说上一小节回答的是“执行模型为什么不能继续停留在单 Worker 黑盒”，那么这一小节回答的就是另一个关键问题：

既然已经进入长任务流水线，那状态和治理到底要做到多细？

这一点建议直接用“四层治理粒度”来理解：

换句话说：

• 主任务级负责“任务定义和总控”
• 批次级负责“租约和执行总览”
• 分片级负责“局部调度和局部恢复”
• 对象级负责“最终状态机和精准补偿”

只有这四层同时存在，系统才能真正从“黑盒长循环”升级成“可治理流水线”。

这里最需要强调的是：supplier_sync_state_ledger 不是可选项，而是工业级流水线的指挥中心。没有它，100w+ 对象同步很容易退化成黑盒。原因非常直接：

• 并发主权：Consumer 通过状态机和 CAS 抢占对象级执行权，避免 MQ 重投后的并发踩脚。
• 对账主权：大盘进度、成功/失败/跳过统计不能去扫 HBase，只能依赖 MySQL 轻量聚合。
• 补偿主权：失败对象必须能被精准捞出做二次重试，而不是把 100w 对象整批重跑。
• 时间差防线：消息在 MQ 里排队数小时后，Consumer 仍可结合状态台账和实时版本做写前保护，阻断旧数据覆盖新数据。

在这套四层模型里，还要继续追问一步：

如果已经接受“分阶段 + 分片流水线”的总体架构，那么主任务表和分片任务表是否可以继续简化，甚至完全去掉？

答案是：可以简化，但不建议完全去掉。

先把三个层级的设计摆出来：

为什么说“可以简化”，是因为不是所有团队都需要一开始就把任务治理做满：

• 如果供应商数量有限、批次规模中等、人工介入极少，确实可以只保留一个轻量 supplier_sync_task 主任务表，把 shard 直接作为内存对象或消息体投到 MQ。
• 如果只是原型验证或单一供应商接入，甚至可以连主任务表都不做，只保留最基本的批次水位和对象级状态。

但为什么又说“不建议完全去掉”，核心原因在于：分片任务表的价值，不只是为了多一张表，而是为了让每个 shard 成为可独立调度、可重试、可监控的单元。

一旦完全去掉 supplier_sync_shard_task，系统通常会退化成下面这种模式：

• Sharder 把 shard 直接打进 MQ
• Fetcher / Publisher 靠对象级 state_ledger 兜底
• 失败后主要依赖日志和对象级状态做恢复

这套模式在小规模下能跑，但会损失几个非常关键的治理能力：

1. 无法方便地对某个分片整体暂停、重跑或摘除

   你可以重试单个对象，但很难回答“第 42 片整体是不是应该停下来人工排查”。

2. 分片级进度、失败率和耗时统计会明显变难

   主任务级只能看到大盘，对象级能看到细粒度结果，但中间缺失了最有价值的“分片级执行视角”。

3. Sharder 失败后的恢复逻辑会更绕

   如果没有持久化的分片任务表，很多恢复动作都会退化成“重新算一遍 shard，再试图和现有 MQ / ledger 对齐”，实现和排障复杂度都会上升。

所以更稳的结论应该是：

• 生产级默认推荐完整版：主任务表 + 分片任务表都保留。
• 中等规模可以接受简化版：保留轻量主任务表，去掉分片任务表，但要明确接受分片级治理能力下降。
• 极简版只适合原型或非常小的单任务系统：它能跑通链路，但不应该作为长期演进目标。

如果把这一判断和上面的方案 A / B 放在一起看，会更清楚：

• 方案 A 主要是在执行策略上更接近“单 Worker 串行”。
• 方案 B 是在架构分层上进入“分阶段 + 分片流水线”。
• 而“主任务表 / 分片任务表是否完整保留”，是在方案 B 内部进一步决定治理能力要做到什么层级。

也就是说，执行模型 和 治理建模粒度 是两个相关但不同的决策，不应该混成一件事。

3.6.2.4 四阶段的关键设计摘要

这张表的作用不是替代后文细节，而是先把最关键的五个工程判断压缩成“高价值摘要”。

3.6.2.5 Phase 0 ~ 3 的一句话设计判断

如果把四阶段模型再压缩一层，每个 Phase 最重要的判断其实只有下面几句：

• Sharder：决定同步边界，支持策略可插拔，并允许慢源退化成单 shard 串行模式。
• Fetcher：负责忠实搬运和 checkpoint 推进，强调页级批量物化，不承担最终业务裁决。
• Transformer：只做外部语义到平台语义的翻译、标准化和风险前移，不越权决定是否正式发布。
• Publisher：对象级状态主权、最终版本保护和正式落地都收口在这一层，不能绕过商品主域写侧。

这一小节的目标不是预演后文，而是帮读者先建立“每个 Phase 最值钱的一句话判断”。

3.6.3 整体设计：任务治理

这一节开始，讨论重点会从“怎么设计执行链路”进一步切到“怎么治理长任务”。如果说前面的内容主要在回答架构和建模，那么这里开始回答的就是另一个更难的问题：

当任务已经进入真实生产环境，开始面对慢源、脏数据、长尾失败、版本冲突和人工介入需求时，系统靠什么机制把风险兜住。

3.6.3.1 异常治理：Shard 失败、补偿与人工介入

对于某酒店供应商这类长任务，最忌讳的一件事就是把异常恢复逻辑塞回主同步链路，最后把 Shard -> Fetch -> Transform -> Publish 重新写成一个满是分支的大黑盒。更稳的做法是：

1. 主同步任务负责正常生产和推进 shard。
2. shard 失败时，只更新主表状态和 checkpoint，并写失败证据。
3. 专门的 supplier_sync_error_task 负责消费失败 shard，判断是自动重试、人工介入、跳过归档，还是柔性补偿。

这里的关键分工是：

• supplier_sync_shard_task：恢复主权表，记录“当前在哪个阶段、从哪恢复”。
• supplier_sync_shard_error_log：失败审计表，记录“为什么失败、失败时上下文是什么”。
• supplier_sync_error_task：异常治理表，记录“这次准备怎么处理失败 shard、处理结果如何”。

这样设计的好处是：

• 主同步任务保持干净，不被异常治理分支污染。
• 大批量任务不会因为少量坏 shard 长时间卡死。
• 失败治理从“看日志人工排查”升级成“显式任务化治理”。

这里需要特别区分两类失败：

• 分片级失败：例如整个页拉取失败、某个 shard 持续超时、某片需要整体暂停。
• 对象级失败：例如某个酒店映射失败、某个对象版本冲突、某条记录被跳过。

分片级失败强调“任务管理和恢复入口”，对象级失败强调“状态机和精准补偿”，两者不能混成一层。

3.6.3.2 收尾、对账与批次终结

长任务真正困难的地方，不只是把对象拉回来、发出去，还在于最后怎么安全收口。

收尾阶段至少要解决两件事：

1. 这次批次什么时候算真正结束
2. 全量 List 模式下，哪些对象本次缺席，应不应该自动下线

先看批次终结判定。推荐的收口协议是：

• 接入层用 INCRBY + end_of_stream + DECR 的动态计数模型推进
• 大盘统计基于 supplier_sync_state_ledger 做轻量聚合
• 如果 Consumer 猝死或 Redis 计数异常，再由 DB 旁路盘点补触发归档

这意味着批次收尾不是“看最后一页拉完了没”，而是要同时看：

• end_of_stream=true
• 对象级消费是否已经收敛
• 大盘是否进入可归档状态

再看全量 List 下的自动下线。它本质上不是行级逻辑，而是批次级治理动作：

• 批次启动时记录 T_start
• 本批次命中的对象刷新最近打卡时间
• 批次收尾时，找出仍在线但最近打卡时间早于 T_start 的对象，形成候选下线集
• 先做比例熔断，再决定是否通过统一治理链路发起下线

这里的关键判断只有两个：

• 全量下线识别必须放在批次收尾，而不是行级消费里顺手判断
• 候选下线集必须先做比例熔断和原因归因，不能直接批量改正式状态

3.6.3.3 本节结论：为什么这是一条工业级长任务流水线

把 3.6.2 和 3.6.3 合在一起看，这条链路之所以值得单独建模，不是因为它“多了几个任务表”，而是因为它同时满足了工业级长任务系统的几个关键特征：

• 它不是“大 Consumer 泥球”，而是显式分成 Sharder / Fetcher / Transformer / Publisher 四个职责边界。
• 它不是“单 Worker 串行黑盒”，而是在总体架构上进入了可观测、可恢复、可补偿的流水线模式。
• 它不是“只会拉数据”，而是把主任务、批次、分片、对象四层治理粒度都建了出来。
• 它不是“跑完就结束”，而是把异常治理、批次终结、自动下线和正式发布保护都纳入了主设计。

换句话说，3.6.2 讲的是整体设计和关键决策，3.6.3 讲的是任务治理；两者合在一起，才构成这类供应商同步真正的工程主体。

所以这条链路的本质，不是一条“供应商拉取脚本”，而是一条 分阶段、强状态、可恢复、可治理 的工业级同步流水线。

3.6.4 Phase 0: Sharder 设计详情

Sharder 不处理具体酒店数据，它只负责回答两个问题：

1. 这次同步的边界是什么。
2. 这次同步应该切成多大的独立执行单元。

输入通常是主任务元数据，例如：

• supplier_id=hotel_supplier_a
• sync_mode=FULL / DELTA
• trigger_time
• task_code

输出是一个 ShardContext 列表或流。每个 ShardContext 至少需要包含：

• batch_id
• shard_id
• sync_mode
• fetch_cursor / page_range / hotel_id_list / geo_scope
• checkpoint
• priority

一个统一的 ShardContext 可以抽象成：

{
  "batch_id": "sb_20260604_0001",
  "shard_id": "hotel-supplier-delta-023",
  "supplier_id": "hotel_supplier_a",
  "sync_mode": "DELTA",
  "hotel_id_list": ["HOTEL_EXT_78231", "HOTEL_EXT_78232"],
  "fetch_cursor": null,
  "page_range": null,
  "checkpoint": null,
  "priority": 50
}

Sharder 最核心的价值是可插拔策略。典型策略如下：

对某酒店供应商这类慢源，虽然可以在概念上保留 Sharder，但很多时候真正落地会退化成：

• 全量同步时，只生成少量大 shard，甚至只有 1 个串行 shard。
• 增量同步时，先通过 Change Log 拿到过去一段时间变化的酒店 ID，再按每 500 家一包切成多个 shard。

在工程落地上，Phase 0: Sharder 至少需要两张表来承接主任务和分片任务：

这里的职责边界要非常明确：

1. supplier_sync_task 不是纯静态模板，它既回答“应该怎么跑”，也回答“这类任务现在整体跑得怎么样”。
2. supplier_sync_shard_task 也不是单纯技术分片，它既回答“这次具体切出了什么”，也回答“哪一片卡住了、哪一片失败了、哪一片该补偿”。

可以把它理解成：

• supplier_sync_task 是某酒店供应商同步这类任务的“作战计划模板 + 总控台”。
• supplier_sync_shard_task 是 2026-06-04 这次增量同步真正拆出来的 100 个 shard 任务 + 分片级执行控制台。

这两张表之所以必须同时承担 任务管理，核心原因只有一个：避免任务黑盒。如果它们只保存配置和少量执行元数据，那么一旦同步跑了 6 小时、卡在第 42 个 shard、某个供应商接口反复超时，平台只能看到“任务还没结束”，却不知道：

• 当前主任务是卡在分片生成、数据拉取、转换还是发布。
• 哪些 shard 已完成，哪些 shard 在重试，哪些 shard 已经失败待人工处理。
• 是否需要整体暂停主任务，还是只重试个别 shard。
• 当前失败是否是供应商全局故障，还是局部数据脏点。

所以更准确的表达应该是：

supplier_sync_task 负责主任务级管理与可观测，supplier_sync_shard_task 负责 shard 级管理与可定位。两者合起来，才能把超长同步任务从黑盒变成可治理流水线。

具体来说，建议它们共同承担以下管理能力：

下面给一组简化样例数据。

主任务样例：supplier_sync_task

{
  "id": 2001,
  "task_code": "HOTEL_SUPPLIER_DELTA_SYNC",
  "supplier_id": 88,
  "sync_mode": "PULL",
  "data_scope": "HOTEL",
  "sharder_type": "DELTA_SHARDER",
  "shard_strategy_version": "v3",
  "concurrency_policy": "SERIAL_FETCH_PARALLEL_PUBLISH",
  "status": "ACTIVE",
  "cron_expr": "0 */1 * * * ?",
  "last_run_at": "2026-06-04 01:00:00",
  "next_run_at": "2026-06-04 02:00:00",
  "last_batch_id": "sb_20260604_010000",
  "last_run_status": "PARTIAL_SUCCESS",
  "last_stage": "PUBLISHING",
  "ext_json": {
    "delta_window_minutes": 60,
    "shard_size": 500,
    "max_fetch_qps": 2,
    "checkpoint_mode": "CHANGE_LOG_CURSOR"
  }
}

分片任务样例：supplier_sync_shard_task

[
  {
    "id": 910001,
    "batch_id": "sb_20260604_020000",
    "shard_task_id": "sst_20260604_0001",
    "task_code": "HOTEL_SUPPLIER_DELTA_SYNC",
    "supplier_id": 88,
    "shard_no": 1,
    "shard_type": "HOTEL_ID_LIST",
    "shard_payload": {
      "hotel_ids": ["HOTEL_EXT_78231", "HOTEL_EXT_78232", "HOTEL_EXT_78233"],
      "source_change_cursor": "chg_20260604015900_001"
    },
    "worker_id": "fetcher-03",
    "lease_until": "2026-06-04 02:05:00",
    "checkpoint": null,
    "status": "INIT",
    "retry_count": 0
  },
  {
    "id": 910002,
    "batch_id": "sb_20260604_020000",
    "shard_task_id": "sst_20260604_0002",
    "task_code": "HOTEL_SUPPLIER_DELTA_SYNC",
    "supplier_id": 88,
    "shard_no": 2,
    "shard_type": "HOTEL_ID_LIST",
    "shard_payload": {
      "hotel_ids": ["HOTEL_EXT_78731", "HOTEL_EXT_78732", "HOTEL_EXT_78733"],
      "source_change_cursor": "chg_20260604015900_002"
    },
    "worker_id": "fetcher-04",
    "lease_until": "2026-06-04 02:05:00",
    "checkpoint": "hotel_id=HOTEL_EXT_78732",
    "status": "RUNNING",
    "retry_count": 1,
    "last_error_code": "SUPPLIER_TIMEOUT"
  }
]

如果是全量同步，supplier_sync_task 不变，但 supplier_sync_shard_task 的 shard_payload 往往会换成 ID 区间或地理范围，例如：

{
  "batch_id": "sb_20260605_020000",
  "shard_task_id": "sst_20260605_0042",
  "task_code": "HOTEL_SUPPLIER_FULL_SYNC",
  "shard_no": 42,
  "shard_type": "HOTEL_ID_RANGE",
  "shard_payload": {
    "start_hotel_id": 420000,
    "end_hotel_id": 429999,
    "country_code": "TH"
  },
  "checkpoint": "hotel_id=424120",
  "status": "RUNNING",
  "retry_count": 1
}

3.6.5 Phase 1: Fetcher 设计详情

Fetcher 接收 ShardContext 后，真正去供应商侧拉数据。它的职责非常克制：

• 调用供应商 API
• 处理鉴权、限流、超时、重试
• 维护当前 shard 的 current_checkpoint
• 把原始响应保存成 RAW snapshot

它不负责做平台业务语义判断，也不负责最终商品裁决。

对某酒店供应商慢源，更推荐 Fetcher 保持串行慢拉：

• 一次只拉一页或一个 cursor
• 每成功一页就推进一次 current_checkpoint
• 每成功一页就落一次 RAW snapshot
• 机器重启后从最近一次安全 checkpoint 继续

为了让后续处理和排障更稳，推荐在 RAW snapshot 旁边同时保存一个轻量 raw_hash。它不是最终业务裁决依据，但很适合用于：

• 辅助判断供应商本次返回内容是否真的变化
• 观测重复页、脏页和异常重放
• 给后续 payload_hash / hash_sign 提供更稳定的底层输入

这一步的本质是：忠实搬运外部数据，并把“已经安全拉到哪里”这件事固化下来。

3.6.6 Phase 2: Transformer 设计详情

Transformer 的输入是 RAW snapshot，输出是内部标准酒店模型。它主要做 4 类工作：

1. 外部字段映射，例如某酒店供应商的酒店、房型、价格日历字段映射到平台内部模型。
2. 数据标准化，例如国家、城市、币种、时区、退款规则统一口径。
3. 基础校验与风险标注，例如字段缺失、模板失配、异常价格、可疑履约规则。
4. 生成平台标准对象快照，供后续 Publisher 使用。

这一步建议只做“把外部语义翻译成平台语义”，不要在这里直接下最终发布裁决。因为：

• Transformer 看得到原始数据，但看不到最终正式商品主权。
• 它适合做结构化解释，不适合做最终线上覆盖裁决。

推荐在标准化产物里额外生成两类辅助信息：

• normalized_hash：标准化后的对象哈希，用于观测和局部去重短路。
• field_version_map：记录本次变更覆盖了哪些字段，供 Publisher 做更细粒度的字段主导权判断。

例如供应商只对价格、房态、上下架状态有主导权，而标题、卖点、运营标签仍以平台人工编辑为准时，field_version_map 能明显降低“整对象覆盖”的粗暴程度。

3.6.7 Phase 3: Publisher 设计详情

Publisher 的输入是标准酒店模型，输出是持久化和分发结果。它要承担的事情最多，但边界也最重要：

• Upsert supplier_sync_state_ledger
• 更新 supplier_product_mapping_tab
• 投递发布消息或调用商品主域写侧
• 根据实时 publish_version、锁定状态、字段主导权做最终写前保护
• 记录 SUCCESS / FAILED / SKIPPED

这一层才是供应商同步与商品主域、审核治理、版本保护真正接壤的地方。换句话说：

Sharder 决定怎么切，Fetcher 决定怎么拉，Transformer 决定怎么翻译，Publisher 决定怎么以平台规则落地。

如果把这 4 步混在一起写成一个大 Consumer，短期能跑，长期几乎一定失控。

在真正进入 Publisher 的对象级落地与主域交互之前，还需要补一个批次级视角。这里的批次级运行状态，主要沉淀在 supplier_sync_batch。它记录的是“这次同步整体怎么跑、现在跑到哪、谁在跑、租约是否有效”，而不是每个酒店对象的明细结果。

一个典型的 supplier_sync_batch 可以长这样：

如果用一条样例记录来表达，可以写成：

{
  "batch_id": "sb_20260604_0001",
  "task_code": "HOTEL_SUPPLIER_PULL_DAILY",
  "status": "RUNNING",
  "sync_batch_version": "20260604-1",
  "trigger_id": "xxljob-20260604-020000",
  "worker_id": "sw-03",
  "lease_token": "lt_8f7a9c21",
  "lease_until": "2026-06-04 02:15:15",
  "heartbeat_at": "2026-06-04 02:10:15",
  "current_checkpoint": "cursor=abc123&page=250",
  "last_checkpoint_at": "2026-06-04 02:09:58",
  "page_no": 250,
  "pulled_count": 125000,
  "enqueued_count": 125000,
  "success_count": 118420,
  "failed_count": 312,
  "skipped_count": 6268,
  "end_of_stream": false,
  "last_error_code": null,
  "last_error_message": null
}

3.6.7.1 Publisher 处理阶段时序图

sequenceDiagram
    participant MQP as "Publish MQ"
    participant CONS as "Publisher Consumer"
    participant HDB as "HBase / Object Store"
    participant SL as "supplier_sync_state_ledger"
    participant P as "商品主域写侧"

    loop 行级消费
        CONS->>MQP: 拉取 item message (获取 outer_goods_id)
        CONS->>HDB: 读取 RAW / NORMALIZED snapshot
        CONS->>SL: CAS 强占状态：INIT -> PROCESSING
        CONS->>P: 实时反查线上正式 DTO 与 publish_version
        alt 运营锁定 / 线上版本已演进
            P-->>CONS: REJECT_STALE_OR_LOCKED
            CONS->>SL: 更新 state_ledger=SKIPPED
        else READY_TO_UPSERT
            P-->>CONS: ready
            CONS->>SL: 更新 state_ledger=READY_TO_UPSERT
        else mapping失败 / 字段缺失 / 高风险阻断
            CONS->>SL: 更新 state_ledger=FAILED
        end
    end

这张图只回答一个问题：Publisher 如何在真正落地前，拦截因为队列时间差导致的旧数据覆盖新数据。

3.6.7.2 Publisher 请求商品主域写侧时序图

sequenceDiagram
    participant CONS as "Publisher Consumer"
    participant P as "商品主域写侧"
    participant SL as "supplier_sync_state_ledger"

    CONS->>SL: 读取 object_key / last_snapshot_ref
    CONS->>P: 携带标准对象 + base_publish_version + ownership_result 发起写请求
    alt 写入成功
        P-->>CONS: UPSERT_ACCEPTED / Draft 已写入
        CONS->>SL: 更新 state_ledger=SUCCESS
    else 线上版本已演进或对象被锁定
        P-->>CONS: SKIPPED_BY_VERSION_OR_LOCK
        CONS->>SL: 更新 state_ledger=SKIPPED
    else 映射失败 / 字段缺失 / 高风险阻断
        P-->>CONS: REJECTED
        CONS->>SL: 更新 state_ledger=FAILED
    end

这张图只回答最后一个问题：Publisher 在完成标准对象准备后，如何把结果请求到商品主域写侧。

3.6.8 表和版本映射

3.6.9 某酒店供应商同步样例数据

下面给一组简化样例，帮助把这些表和某酒店供应商同步场景串起来。这里默认同步对象是 HOTEL_EXT_78231，同步模式是串行 Pull，目标是把供应商侧的酒店基础信息、房型信息和价格/库存差异同步到平台治理链路。

如果继续细化到对象级状态，可以直接在 supplier_sync_state_ledger 上看到这种结果：

3.7 核心功能：库存

3.7.1 场景画像

供给平台里的库存能力，不是库存中心那套账本、预占、确认、释放的权威事实模型，而是 库存 B 端入口与操作编排层。

这一节要覆盖的，不只是“调库存”这种通用动作，更重要的是：创建商品时，如何兼容创建库存。

在酒旅、到店餐饮、景区票务和本地生活券这类场景里，创建库存至少要支持 3 种完全不同的模式：

1. QTY：纯数字库存

   • 创建商品时，运营手工填写库存数字，例如 100
   • 供给平台只需要声明初始数量
   • 库存中心持有真实可售数量
   • 下单时扣减的是数字，不是预先存在的券码

2. SYS_GEN：系统动态发券

   • 创建商品时，同样填写一个库存数字，例如 500
   • 但用户支付成功后，平台会按规则动态生成核销券码或二维码
   • 库存本体仍然是数量，券码只是履约载体
   • 除了初始数量外，还要声明券码生成规则或生成策略

3. EXT_POOL：外部死码池

   • 创建商品时，不能把“手填数量”当成库存真相
   • 运营必须上传 Excel 或外部批次文件，文件内包含真实券码、密码、有效期等信息
   • 库存的真正来源是券码池里的有效行数
   • 用户下单时不是“数字减一”，而是从券码池中占用一张真实的死码

因此，供给平台在“创建商品”时，必须同步声明一项关键业务属性：

• inventory_mode = QTY / SYS_GEN / EXT_POOL

商品创建成功后，供给平台再根据这个模式，把库存初始化动作路由到不同的命令编排路径，而不是所有商品都走同一种 CreateInventory。

这一节后续要覆盖的典型动作包括：

• 初始化库存
• 调整库存
• 导入券码
• 生码
• 锁库存

核心边界要先钉死：

入口归供给，事实归库存。

也就是说，供给平台负责承接操作入口、审批、权限、审计、错误文件和命令编排，但不持有库存事实、预占、账本或券码池权威状态。

3.7.2 创建商品时顺带创建库存的时序图

sequenceDiagram
    participant U as "运营/商家"
    participant W as "供给平台 Web"
    participant T as "product_supply_task"
    participant I as "product_supply_task_item"
    participant S as "供给平台服务"
    participant P as "商品主域写侧"
    participant IC as "库存中心"
    participant F as "错误文件/审计"

    U->>W: 创建商品时选择 inventory_mode 并填写库存参数
    W->>T: 创建 task
    W->>I: 创建 task_item
    W->>S: 提交商品与库存配置
    S->>T: task=CREATED
    S->>I: item=CREATED
    S->>S: 校验商品参数 / inventory_mode / 权限 / 参数完整性
    alt 校验失败
        S->>I: 更新 item=FAILED，记录 error_code / error_message
        S->>T: 更新 task=FAILED
        S->>F: 记录错误文件 / 审计日志
    else 校验通过
        S->>T: task=CREATING_PRODUCT
        S->>P: CreateDraft / SubmitCreate
        alt 商品主域写侧失败
            P-->>S: 返回 error_code / error_message
            S->>I: 更新 item=FAILED
            S->>T: 更新 task=FAILED
            S->>F: 记录错误文件 / 审计日志
        else 商品主域写侧成功
            P-->>S: 返回 item_id / draft_id
            S->>T: task=CREATING_INVENTORY
            S->>S: 根据 inventory_mode 路由库存创建命令
            alt QTY 纯数字库存
                S->>IC: CreateInventory(item_id, init_quantity)
            else SYS_GEN 系统动态发券
                S->>IC: CreateInventory(item_id, init_quantity, voucher_rule)
            else EXT_POOL 外部死码池
                S->>IC: CreateInventory(item_id, pool_mode=EXT_POOL)
                S->>IC: ImportCodeBatch(item_id, code_batch_file / code_rows)
            end
            alt 库存中心成功
                IC-->>S: 返回 command_id / result
                S->>I: 更新 item=DONE
                S->>T: 更新 task=DONE
            else 库存中心失败
                IC-->>S: 返回 error_code / error_message
                S->>I: 更新 item=FAILED
                S->>T: 更新 task=INVENTORY_FAILED
                S->>F: 记录错误文件 / 审计日志
            end
        end
    end

这张图只回答一个问题：

当“创建商品”和“创建库存”被收敛到同一个供给任务里时，供给中心如何先完成商品创建，再按库存类型路由库存初始化。

因此它故意不展开库存中心内部怎么记账、怎么预占、怎么维护券码池，只强调供给中心这一侧的 4 个动作：

• 商品创建和库存初始化共用一个任务锚点
• 商品创建阶段必须先声明 inventory_mode
• 供给中心先调商品主域写侧创建商品，再根据结果决定是否进入库存初始化
• 供给中心先把库存初始化操作任务化、审计化
• 供给中心根据 inventory_mode 决定库存初始化走哪条命令路径
• QTY 和 SYS_GEN 都以“数量”为库存源头
• EXT_POOL 以“券码池”为库存源头，手填数量不是最终真相

3.7.3 商品已存在时单独创建或增加库存的时序图

如果商品已经创建完成，后续还需要支持一种更轻量的库存入口：单独创建库存或增加库存。这时不再需要走商品创建链路，而是直接围绕既有 item_id 发起库存操作。

sequenceDiagram
    participant U as "运营/商家"
    participant W as "供给平台 Web"
    participant T as "product_supply_task"
    participant I as "product_supply_task_item"
    participant S as "供给平台服务"
    participant IC as "库存中心"
    participant F as "错误文件/审计"

    U->>W: 基于已有 item_id 发起创建库存 / 增加库存
    W->>T: 创建 task
    W->>I: 创建 task_item
    W->>S: 提交 item_id / inventory_mode / inventory_action
    S->>T: task=CREATING_INVENTORY
    S->>I: item=CREATED
    S->>S: 校验 item_id / inventory_mode / 权限 / 参数完整性
    alt 校验失败
        S->>I: 更新 item=FAILED
        S->>T: 更新 task=FAILED
        S->>F: 记录错误文件 / 审计日志
    else 校验通过
        alt 创建或增加数量库存
            S->>IC: CreateInventory / AdjustInventory(item_id, quantity)
        else 创建或增加系统生券库存
            S->>IC: CreateInventory / AdjustInventory(item_id, quantity, voucher_rule)
        else 导入外部死码池
            S->>IC: CreateInventory(item_id, pool_mode=EXT_POOL)
            S->>IC: ImportCodeBatch(item_id, code_batch_file / code_rows)
        end
        alt 库存中心成功
            IC-->>S: 返回 command_id / result
            S->>I: 更新 item=DONE
            S->>T: 更新 task=DONE
        else 库存中心失败
            IC-->>S: 返回 error_code / error_message
            S->>I: 更新 item=FAILED
            S->>T: 更新 task=INVENTORY_FAILED
            S->>F: 记录错误文件 / 审计日志
        end
    end

这张补充图只回答另一个问题：

当商品已经存在时，供给中心如何围绕既有 item_id 单独创建库存或增加库存。

它和前一张图的区别在于：

• 不再调用商品主域写侧创建商品
• 任务直接从 CREATING_INVENTORY 开始
• 入口重点变成已有商品的库存补建、补量、导码和补码

3.7.4 创建商品与创建库存共用任务的状态机

当商品创建和库存初始化被收敛到同一个 product_supply_task 时，任务状态机要同时表达两段动作：

• 商品是否已经成功创建
• 库存是否已经成功初始化

推荐的任务级状态如下：

如果希望进一步表达行级明细，product_supply_task_item 可以保持轻量：

这个状态机的关键点不是追求状态越多越好，而是要明确区分两类失败：

• 商品没建成：FAILED
• 商品建成了，但库存没建成：INVENTORY_FAILED

只有这样，运营侧才能知道是整单回退，还是只需要补库存。

3.7.5 关键设计点

如果把这一节再往上提炼，供给平台在“商品创建 + 库存创建”这条链路上，真正要解决的是 4 个高风险技术点：

1. 多模式库存建模怎么统一入口、分离事实。
2. 商品创建和库存初始化如何用一个联合任务表达半成功。
3. 单品页面直接改库存时，为什么不能把绝对值直接写回库存中心。
4. 外部死码池为什么必须走“资源导入”，而不是“数量加减”。

3.7.6 表和版本映射

这一节刻意不再把库存入口拆成太多请求对象。对于供给平台来说，最重要的是：

• 任务锚点是否存在
• 任务明细是否能表达失败和补偿
• 商品创建时是否声明了正确的 inventory_mode

像 base_inventory_version、导码文件引用、券码规则等更细的字段，继续在 3.7.5 关键设计点 和具体时序图里表达即可，不必在这里再拆成一串对象清单。

3.8 发布编排、商品主域交互与外部系统集成

供给平台在这一层不做正式落库，而是负责把治理结果编排成可执行的发布动作，并把结果扩散给商品主域和外部协同系统。

建议把这一节拆成三类内容来讲：

1. 商品主域交互

   • 发布命令
   • base_publish_version
   • product_publish_record
   • Draft / Staging / QC / merge 边界

2. 外部系统集成

   • 搜索
   • 缓存
   • 计价
   • 营销
   • 数据平台 / 画像 / 订阅方

3. 一致性表达

   • 供给平台不做正式落库
   • 供给平台负责发布编排和跟踪
   • 商品主域负责正式 merge
   • 外部系统通过事件 / Outbox / 投影刷新感知

这里要钉死一句：

供给平台负责编排和触发发布；Draft / Staging / QC / publish merge 由商品主域写侧闭环完成；搜索、缓存、计价、营销和数据平台等外部系统通过投影刷新或事件订阅感知变更。

3.9 异常处理与运营闭环

供给平台最终要把失败运营化，而不是把问题藏在日志里。

这里要先钉死一条异常治理总原则：

失败后不能只靠“直接重试覆盖”来掩盖问题，而应该保留完整的审计链路、错误归因和补偿闭环。
只有这样，系统才具备可回放、可追责、可人工修复的工程确定性。

这一小节建议覆盖：

• 错误文件
• 部分成功
• DLQ
• 补偿任务
• 人工修复
• 审计日志
• 质量看板
• 同步任务卡死
• 下架误推送
• mapping 失效后的人工介入与重试

对运营侧要能回答：

• 哪个任务失败了
• 哪一行失败了
• 为什么失败
• 是交互型导入问题还是系统型同步问题
• 能否下载错误文件修复
• 是否可以重新投递

3.10 供给平台表清单与使用场景

这一小节建议显式列出供给平台权威表，避免和商品主域写侧、库存系统混淆。

3.10.1 每个表的使用场景

这里先回答“这张表为什么存在”。保持当前按职责分类的方式不变，但只讨论每张表的使用场景、负责链路和关键边界；字段和 schema 统一收口到 3.10.2。

A. 接入与任务编排表

B. 标准化与冲突治理补充表

C. 发布编排与追踪表

D. 失败恢复与运营治理表

E. 供应商同步执行层表

这里还要明确反向排除：

• product_supply_draft
• product_supply_staging
• product_qc_review

这些不应列为供给平台权威表，而应归商品主域写侧。

3.10.2 每个表的 Schema

这里再回答“这张表长什么样”。字段口径、推荐索引和 DDL 示例统一放在这一小节，避免和 3.10.1 的用途说明重复。

先看最核心的接入与任务编排表。

这 3 张表建议这样设计：

product_supply_task

product_supply_task_item

product_supply_operation_log

推荐索引：

• product_supply_task
  • 唯一索引：uk_receipt_id
  • 普通索引：idx_operation_id
  • 普通索引：idx_source_type_status_created_at
• product_supply_task_item
  • 普通索引：idx_task_id_status
  • 普通索引：idx_object_key
  • 普通索引：idx_publish_record_id
• product_supply_operation_log
  • 普通索引：idx_operation_id_created_at
  • 普通索引：idx_task_id_event_type
  • 普通索引：idx_trace_id

核心表 Schema 示例

前面的表清单和字段说明，解决的是“这张表负责什么”。如果要真正指导团队落库实现，还需要再往下一层，明确这些核心表的 schema 轮廓。

这里不追求一开始就给出百分之百完整的线上 DDL，而是给出一版 足够指导实现、且能稳定支撑当前设计心智 的 schema 示例。团队在落地时可以按业务规模再补分库分表键、审计字段和冷热分层策略。

product_supply_task

CREATE TABLE `product_supply_task` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '任务主键ID',
  `task_no` VARCHAR(64) NOT NULL COMMENT '对外任务号',
  `receipt_id` VARCHAR(64) NOT NULL COMMENT '受理号',
  `trigger_id` VARCHAR(64) DEFAULT NULL COMMENT '触发源ID',
  `operation_id` VARCHAR(64) NOT NULL COMMENT '操作链路ID',
  `task_type` VARCHAR(32) NOT NULL COMMENT 'CREATE / EDIT / IMPORT / API_PUSH / BATCH_EDIT',
  `source_type` VARCHAR(32) NOT NULL COMMENT 'LOCAL / EXCEL / API / ISV / OPS',
  `biz_scope` VARCHAR(32) NOT NULL COMMENT 'PRODUCT / INVENTORY_ENTRY / PUBLISH',
  `execution_mode` VARCHAR(16) NOT NULL COMMENT 'SYNC / ASYNC',
  `operator_type` VARCHAR(16) NOT NULL COMMENT 'MERCHANT / OPS / SYSTEM / ISV',
  `operator_id` VARCHAR(64) NOT NULL COMMENT '操作人ID',
  `merchant_id` BIGINT DEFAULT NULL COMMENT '商家ID',
  `tenant_id` BIGINT DEFAULT NULL COMMENT '租户ID',
  `status` VARCHAR(32) NOT NULL COMMENT '任务状态',
  `worker_id` VARCHAR(64) DEFAULT NULL COMMENT '持有任务的执行器',
  `lease_token` VARCHAR(64) DEFAULT NULL COMMENT '租约令牌',
  `lease_until` DATETIME(3) DEFAULT NULL COMMENT '租约过期时间',
  `heartbeat_at` DATETIME(3) DEFAULT NULL COMMENT '最后心跳时间',
  `last_heartbeat_stage` VARCHAR(64) DEFAULT NULL COMMENT '最后心跳阶段',
  `parse_checkpoint` VARCHAR(512) DEFAULT NULL COMMENT '解析进度断点',
  `file_url` VARCHAR(512) DEFAULT NULL COMMENT '原始上传文件地址',
  `error_file_url` VARCHAR(512) DEFAULT NULL COMMENT '错题本地址',
  `total_count` INT NOT NULL DEFAULT 0 COMMENT '总对象数',
  `success_count` INT NOT NULL DEFAULT 0 COMMENT '成功数',
  `failed_count` INT NOT NULL DEFAULT 0 COMMENT '失败数',
  `skipped_count` INT NOT NULL DEFAULT 0 COMMENT '跳过数',
  `risk_level` VARCHAR(16) DEFAULT NULL COMMENT 'LOW / MEDIUM / HIGH',
  `ext_json` JSON DEFAULT NULL COMMENT '扩展上下文',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_task_no` (`task_no`),
  UNIQUE KEY `uk_receipt_id` (`receipt_id`),
  KEY `idx_operation_id` (`operation_id`),
  KEY `idx_status_lease` (`status`, `lease_until`),
  KEY `idx_source_type_status_created_at` (`source_type`, `status`, `created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供给平台统一任务主表';

product_supply_task_item

CREATE TABLE `product_supply_task_item` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '行级主键ID',
  `task_id` BIGINT UNSIGNED NOT NULL COMMENT '所属任务ID',
  `item_no` INT NOT NULL COMMENT '行号/序号',
  `object_type` VARCHAR(32) NOT NULL COMMENT 'PRODUCT / SKU / OFFER / INVENTORY_ENTRY',
  `object_key` VARCHAR(128) NOT NULL COMMENT '对象业务键',
  `item_action` VARCHAR(32) NOT NULL COMMENT 'CREATE / UPDATE / UPSERT / DELETE / PUBLISH',
  `raw_ref` VARCHAR(512) DEFAULT NULL COMMENT '原始输入引用',
  `normalized_snapshot` JSON DEFAULT NULL COMMENT '标准化快照',
  `diff_summary` JSON DEFAULT NULL COMMENT '差异摘要',
  `business_fingerprint` VARCHAR(64) DEFAULT NULL COMMENT '业务指纹',
  `base_publish_version` BIGINT DEFAULT NULL COMMENT '更新链路版本锚点',
  `publish_record_id` BIGINT DEFAULT NULL COMMENT '发布记录ID',
  `status` VARCHAR(32) NOT NULL COMMENT '行级状态',
  `error_code` VARCHAR(64) DEFAULT NULL COMMENT '错误码',
  `error_message` VARCHAR(1024) DEFAULT NULL COMMENT '错误信息',
  `retry_count` INT NOT NULL DEFAULT 0 COMMENT '重试次数',
  `ext_json` JSON DEFAULT NULL COMMENT '扩展字段',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  KEY `idx_task_id_status` (`task_id`, `status`),
  KEY `idx_object_key` (`object_key`),
  KEY `idx_publish_record_id` (`publish_record_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供给平台任务行级明细表';

product_supply_operation_log

CREATE TABLE `product_supply_operation_log` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '日志主键ID',
  `operation_id` VARCHAR(64) NOT NULL COMMENT '操作链路ID',
  `task_id` BIGINT UNSIGNED DEFAULT NULL COMMENT '关联任务ID',
  `task_item_id` BIGINT UNSIGNED DEFAULT NULL COMMENT '关联行级ID',
  `event_type` VARCHAR(32) NOT NULL COMMENT 'SUBMIT / RETRY / WITHDRAW / GENERATE_ERROR_FILE / TRIGGER_COMPENSATION / PUBLISH',
  `event_stage` VARCHAR(32) NOT NULL COMMENT 'INGEST / VALIDATE / ROUTE / PUBLISH / COMPENSATE',
  `operator_type` VARCHAR(16) NOT NULL COMMENT 'MERCHANT / OPS / SYSTEM / SCHEDULER',
  `operator_id` VARCHAR(64) NOT NULL COMMENT '触发主体',
  `before_snapshot` JSON DEFAULT NULL COMMENT '变更前摘要',
  `after_snapshot` JSON DEFAULT NULL COMMENT '变更后摘要',
  `result_status` VARCHAR(16) NOT NULL COMMENT 'SUCCESS / FAILED / SKIPPED',
  `result_code` VARCHAR(64) DEFAULT NULL COMMENT '结果码',
  `message` VARCHAR(1024) DEFAULT NULL COMMENT '审计说明',
  `trace_id` VARCHAR(128) DEFAULT NULL COMMENT '链路追踪ID',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '事件时间',
  PRIMARY KEY (`id`),
  KEY `idx_operation_id_created_at` (`operation_id`, `created_at`),
  KEY `idx_task_id_event_type` (`task_id`, `event_type`),
  KEY `idx_trace_id` (`trace_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供给平台操作审计日志表';

supplier_sync_task

CREATE TABLE `supplier_sync_task` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `task_code` VARCHAR(64) NOT NULL COMMENT '业务任务编码',
  `supplier_id` BIGINT NOT NULL COMMENT '供应商ID',
  `sync_mode` VARCHAR(16) NOT NULL COMMENT 'PULL / PUSH',
  `data_scope` VARCHAR(32) NOT NULL COMMENT 'HOTEL / PRODUCT / OFFER',
  `sharder_type` VARCHAR(32) NOT NULL COMMENT 'FULL_SHARDER / DELTA_SHARDER / CURSOR_SHARDER',
  `shard_strategy_version` VARCHAR(32) NOT NULL COMMENT '分片策略版本号',
  `concurrency_policy` VARCHAR(32) NOT NULL COMMENT 'SINGLE_WORKER / SERIAL_BATCH',
  `status` VARCHAR(16) NOT NULL COMMENT 'ACTIVE / PAUSED',
  `cron_expr` VARCHAR(64) DEFAULT NULL COMMENT '调度表达式',
  `last_batch_id` VARCHAR(64) DEFAULT NULL COMMENT '最近一次执行批次',
  `last_run_status` VARCHAR(32) DEFAULT NULL COMMENT '最近一次执行结果',
  `last_stage` VARCHAR(32) DEFAULT NULL COMMENT '最近一次运行停留阶段',
  `last_run_at` DATETIME(3) DEFAULT NULL COMMENT '最近一次运行时间',
  `next_run_at` DATETIME(3) DEFAULT NULL COMMENT '下一次调度时间',
  `ext_json` JSON DEFAULT NULL COMMENT '扩展配置',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_task_code` (`task_code`),
  KEY `idx_supplier_id_status` (`supplier_id`, `status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供应商同步任务定义表';

supplier_sync_batch

CREATE TABLE `supplier_sync_batch` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `batch_id` VARCHAR(64) NOT NULL COMMENT '批次ID',
  `task_code` VARCHAR(64) NOT NULL COMMENT '业务任务编码',
  `supplier_id` BIGINT NOT NULL COMMENT '供应商ID',
  `status` VARCHAR(32) NOT NULL COMMENT 'RUNNING / FAILED / DONE / PARTIAL_SUCCESS',
  `sync_batch_version` VARCHAR(64) NOT NULL COMMENT '批次版本号',
  `trigger_id` VARCHAR(64) DEFAULT NULL COMMENT '触发源ID',
  `worker_id` VARCHAR(64) DEFAULT NULL COMMENT '执行worker',
  `lease_token` VARCHAR(64) DEFAULT NULL COMMENT '租约令牌',
  `lease_until` DATETIME(3) DEFAULT NULL COMMENT '租约到期时间',
  `heartbeat_at` DATETIME(3) DEFAULT NULL COMMENT '最后心跳时间',
  `current_checkpoint` VARCHAR(512) DEFAULT NULL COMMENT '当前checkpoint',
  `last_checkpoint_at` DATETIME(3) DEFAULT NULL COMMENT '最近推进checkpoint时间',
  `page_no` INT NOT NULL DEFAULT 0 COMMENT '当前页号',
  `pulled_count` BIGINT NOT NULL DEFAULT 0 COMMENT '已拉取数量',
  `enqueued_count` BIGINT NOT NULL DEFAULT 0 COMMENT '已投MQ数量',
  `success_count` BIGINT NOT NULL DEFAULT 0 COMMENT '成功数',
  `failed_count` BIGINT NOT NULL DEFAULT 0 COMMENT '失败数',
  `skipped_count` BIGINT NOT NULL DEFAULT 0 COMMENT '跳过数',
  `end_of_stream` TINYINT(1) NOT NULL DEFAULT 0 COMMENT '是否拉到尾页',
  `last_error_code` VARCHAR(64) DEFAULT NULL COMMENT '最近错误码',
  `last_error_message` VARCHAR(1024) DEFAULT NULL COMMENT '最近错误信息',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_batch_id` (`batch_id`),
  KEY `idx_task_code_status` (`task_code`, `status`),
  KEY `idx_status_lease` (`status`, `lease_until`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供应商同步批次表';

supplier_sync_shard_task

CREATE TABLE `supplier_sync_shard_task` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `batch_id` VARCHAR(64) NOT NULL COMMENT '所属批次ID',
  `shard_task_id` VARCHAR(64) NOT NULL COMMENT '分片任务ID',
  `task_code` VARCHAR(64) NOT NULL COMMENT '业务任务编码',
  `supplier_id` BIGINT NOT NULL COMMENT '供应商ID',
  `shard_no` INT NOT NULL COMMENT '分片序号',
  `shard_type` VARCHAR(32) NOT NULL COMMENT 'HOTEL_ID_LIST / HOTEL_ID_RANGE / GEO_SCOPE / CURSOR',
  `shard_payload` JSON NOT NULL COMMENT '分片上下文，如酒店ID列表、ID范围、国家城市、cursor等',
  `status` VARCHAR(32) NOT NULL COMMENT 'INIT / RUNNING / SUCCESS / FAILED / PARTIAL_SUCCESS / SKIPPED / PAUSED',
  `checkpoint` VARCHAR(512) DEFAULT NULL COMMENT '当前分片checkpoint',
  `retry_count` INT NOT NULL DEFAULT 0 COMMENT '重试次数',
  `worker_id` VARCHAR(64) DEFAULT NULL COMMENT '当前执行worker',
  `lease_token` VARCHAR(64) DEFAULT NULL COMMENT '租约令牌',
  `lease_until` DATETIME(3) DEFAULT NULL COMMENT '租约到期时间',
  `heartbeat_at` DATETIME(3) DEFAULT NULL COMMENT '最后心跳时间',
  `pulled_count` BIGINT NOT NULL DEFAULT 0 COMMENT '该分片已拉取数量',
  `published_count` BIGINT NOT NULL DEFAULT 0 COMMENT '该分片已发布数量',
  `failed_count` BIGINT NOT NULL DEFAULT 0 COMMENT '该分片失败数量',
  `last_error_code` VARCHAR(64) DEFAULT NULL COMMENT '最近错误码',
  `last_error_message` VARCHAR(1024) DEFAULT NULL COMMENT '最近错误信息',
  `ext_json` JSON DEFAULT NULL COMMENT '扩展上下文',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_shard_task_id` (`shard_task_id`),
  KEY `idx_batch_id_status` (`batch_id`, `status`),
  KEY `idx_task_code_batch_id` (`task_code`, `batch_id`),
  KEY `idx_status_lease` (`status`, `lease_until`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供应商同步分片任务表';

supplier_sync_error_task

CREATE TABLE `supplier_sync_error_task` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `error_task_id` VARCHAR(64) NOT NULL COMMENT '错误处理任务ID',
  `batch_id` VARCHAR(64) NOT NULL COMMENT '所属批次ID',
  `source_shard_task_id` VARCHAR(64) NOT NULL COMMENT '来源分片任务ID',
  `recovery_type` VARCHAR(32) NOT NULL COMMENT 'RETRY_FETCH / RETRY_TRANSFORM / RETRY_PUBLISH / MANUAL_REVIEW / SKIP',
  `target_stage` VARCHAR(32) NOT NULL COMMENT 'FETCH / TRANSFORM / PUBLISH',
  `status` VARCHAR(32) NOT NULL COMMENT 'INIT / RUNNING / SUCCESS / FAILED / CANCELED',
  `start_checkpoint` VARCHAR(512) DEFAULT NULL COMMENT '恢复起点',
  `retry_count` INT NOT NULL DEFAULT 0 COMMENT '重试次数',
  `operator_type` VARCHAR(16) DEFAULT NULL COMMENT 'SYSTEM / OPS / SCHEDULER',
  `operator_id` VARCHAR(64) DEFAULT NULL COMMENT '触发主体',
  `last_error_code` VARCHAR(64) DEFAULT NULL COMMENT '最近错误码',
  `last_error_message` VARCHAR(1024) DEFAULT NULL COMMENT '最近错误信息',
  `ext_json` JSON DEFAULT NULL COMMENT '扩展上下文',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_error_task_id` (`error_task_id`),
  KEY `idx_source_shard_task_id_status` (`source_shard_task_id`, `status`),
  KEY `idx_batch_id_status` (`batch_id`, `status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供应商同步错误处理任务表';

supplier_sync_shard_error_log

CREATE TABLE `supplier_sync_shard_error_log` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `batch_id` VARCHAR(64) NOT NULL COMMENT '所属批次ID',
  `shard_task_id` VARCHAR(64) NOT NULL COMMENT '分片任务ID',
  `stage` VARCHAR(32) NOT NULL COMMENT 'FETCH / TRANSFORM / PUBLISH',
  `error_code` VARCHAR(64) NOT NULL COMMENT '错误码',
  `error_message` VARCHAR(1024) DEFAULT NULL COMMENT '错误信息',
  `checkpoint` VARCHAR(512) DEFAULT NULL COMMENT '失败时checkpoint',
  `raw_ref` VARCHAR(512) DEFAULT NULL COMMENT '原始快照引用',
  `normalized_ref` VARCHAR(512) DEFAULT NULL COMMENT '标准化快照引用',
  `retry_count` INT NOT NULL DEFAULT 0 COMMENT '失败时重试次数',
  `trace_id` VARCHAR(128) DEFAULT NULL COMMENT '链路追踪ID',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_shard_task_id_created_at` (`shard_task_id`, `created_at`),
  KEY `idx_batch_id_stage` (`batch_id`, `stage`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供应商同步分片失败日志表';

supplier_sync_state_ledger

CREATE TABLE `supplier_sync_state_ledger` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `supplier_id` BIGINT NOT NULL COMMENT '供应商ID',
  `outer_goods_id` VARCHAR(128) NOT NULL COMMENT '外部对象ID',
  `last_batch_id` VARCHAR(64) DEFAULT NULL COMMENT '最近处理批次',
  `state` VARCHAR(32) NOT NULL COMMENT 'INIT / PROCESSING / SUCCESS / FAILED / SKIPPED / BLOCKED / OFFLINE_CANDIDATE',
  `last_snapshot_ref` VARCHAR(512) DEFAULT NULL COMMENT '最近快照引用',
  `raw_hash` VARCHAR(128) DEFAULT NULL COMMENT '最近原始快照哈希',
  `normalized_hash` VARCHAR(128) DEFAULT NULL COMMENT '最近标准化快照哈希',
  `payload_hash` VARCHAR(64) DEFAULT NULL COMMENT '供应商报文哈希，仅用于观测或安全短路',
  `last_seen_at` DATETIME(3) DEFAULT NULL COMMENT '最近打卡时间',
  `off_reason` VARCHAR(128) DEFAULT NULL COMMENT '下线原因',
  `off_batch_id` VARCHAR(64) DEFAULT NULL COMMENT '候选下线批次',
  `error_code` VARCHAR(64) DEFAULT NULL COMMENT '最近错误码',
  `error_message` VARCHAR(1024) DEFAULT NULL COMMENT '最近错误信息',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_supplier_outer_goods` (`supplier_id`, `outer_goods_id`),
  KEY `idx_last_batch_id_state` (`last_batch_id`, `state`),
  KEY `idx_supplier_state_seen` (`supplier_id`, `state`, `last_seen_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='供应商同步状态台账表';

这些 schema 的核心价值不在于字段有多全，而在于它们把供给平台真正承担的三类主权表达清楚了：

• 任务编排主权：task / task_item
• 审计与补偿主权：operation_log
• 供应商长任务状态主权：supplier_sync_task / batch / state_ledger

其中有两个实现细节需要明确：

• Local 单品创建/编辑虽然是同步交互，但后端仍建议创建轻量 product_supply_task(total_count=1, execution_mode=SYNC)，这样才能统一审计、发布追踪和失败补偿。
• product_supply_task_item.normalized_snapshot 保存的是供给平台完成标准化后的对象快照，用于校验、Diff 和发布编排；它不是商品主域写侧的 Draft / Staging 资产本体。

4. 详细设计 - 商品中心

4.1 商品中心定位与职责边界

商品中心不是“后台录入页面直接写的数据库”，而是 商品主域写侧 + 正式读模型 的组合体。

它负责：

• Draft / Item / Item Snapshot / Merge
• 正式商品的版本演进与发布
• 订单可解释的交易快照
• 面向搜索、缓存、计价、营销、数据平台的正式事件输出

它不负责：

• 供给入口线程池、Excel 跑批、供应商接入
• 库存账本、预占释放、券码池
• 人工审核工作台和审核队列

在这套架构里：

• 供给平台负责接入、标准化、任务编排和发布触发
• QC 平台负责机审核、人工审核和 verdict
• 库存中心负责库存事实和扣减账本
• 商品中心负责把“一个可编辑草稿”收敛成“一个可交易的正式版本”

一句话说，商品中心沉淀的不是录入过程，而是 正式交易契约与其可发布版本。

4.2 决策点

4.2.1 决策点 1：为什么正式商品表不能直接承接后台录入流程

结论：

• 商品中心必须显式拆开 Draft / Item
• 正式表只表达正式商品，不承接 DRAFT / QC_PENDING / REJECTED 这类流程态

4.2.2 决策点 2：只有一个 Draft，提交审核后锁定不允许编辑，是否还需要 Staging

结论：

• 当前设计只保留一个 Draft
• 提交审核后，将 Draft 切换到 QC_PENDING 并禁止继续编辑，用“审核态锁定”保证审的是 A，发的也是 A
• 这个方案本质上是“用运营阻塞换系统简单”，适合审核时间短、运营冲突低、商品结构相对稳定的阶段
• 若未来出现长审核周期、并行准备多个未来版本、定时生效或高确定性版本回滚要求，再演进为 Draft + Staging + Item

4.2.3 决策点 3：商品中心是否需要 base_publish_version

结论：

• 编辑链路必须携带 base_publish_version
• 商品中心是最终的版本冲突裁决者

4.2.4 决策点 4：publish_version 是否需要单独成表

结论：

• 当前商品中心先采用轻量方案，在 product_item_tab 上只保留一个当前 publish_version 数字
• 只要能满足 base_publish_version 校验、并发保护和正式态版本推进，就不必先引入独立版本历史表
• 当后续需要完整发布历史、版本审计、版本级回滚或版本级外部对账时，再平滑演进为独立 publish_version 表

4.2.5 决策点 5：发布时是否允许跨服务长事务

结论：

• 商品中心内部本地事务完成 merge
• 外部系统通过 Outbox / 事件驱动刷新投影

4.2.6 决策点 6：无效变更过滤应该放在供给侧还是商品中心

结论：

• 最终业务级 Diff、幂等和真相裁决由商品中心承担
• 供给侧只做安全过滤，不做最终业务跳过裁决

4.2.7 决策点 7：重复商品识别和供应商无效更新过滤，是否应该前移到供给侧

结论：

• 重复商品识别 和 最终无效更新过滤 都收敛到商品中心
• 供给侧只做拓扑级安全过滤，例如同批次重复页、重投踩脚、无映射死链
• 商品中心负责：
  • 基于业务自然键识别重复商品
  • 基于正式 DTO、字段主导权和 base_publish_version 过滤掉无效更新
  • 基于乐观锁阻断并发覆盖

4.2.8 决策点 8：商品中心写模型的幂等，应该靠单点防重还是多层立体防御

结论：

• 商品中心写模型幂等不能靠单一“防重表”解决
• 在当前 单 Draft + product_item_tab.publish_version + 本地 merge 的设计中，应采用四层立体防御：
  • 请求级幂等：request_id / operation_id
  • 对象级幂等：业务自然键唯一约束
  • 版本级幂等：draft.base_publish_version vs item.publish_version
  • 状态机幂等：状态推进必须条件更新
• 这四层分别拦截：
  • 网络重试 / MQ 重投
  • 重复商品创建
  • 旧请求覆盖新版本
  • 状态乱序推进与僵尸回调

4.2.9 决策点 9：异构异质多品类（服务、时空资源、契约）商品应该如何组织

这个决策点本质上是在权衡：

• 商品中心是否要把话费充值、账单还款、酒店、电影、演出票、优惠券都建成同一种“重商品对象”
• 还是只维护统一交易货架，把真正复杂的物理资源、时空资源和渠道契约下沉到垂直业务域

这里需要先看清 4 个核心冲突：

1. 资产属性冲突
   • 实物、券码更接近静态实体
   • 酒店、电影属于时空资源，价格和可售性跟日期、场次、房型强绑定
   • 话费、账单更接近 API 驱动的数据服务，本身并没有稳定的“自建库存商品”
2. 写频率冲突
   • 实物标题、图文、卖点更新频率低
   • 酒店房态价、话费渠道进价高频波动
   • 如果全部走标准 Draft -> QC -> Publish，正式版本号会被高频资源波动打爆
3. 状态主权冲突
   • 平台是否允许卖某个品类是一层主权
   • 某一家酒店、某一个电影场次、某一个充值通道是否可卖是另一层主权
   • 如果所有资源都塞进商品中心统一状态机，写模型会过度膨胀
4. 交易底座共用需求
   • 尽管供给模型差异极大，但购物车、营销、支付、收银台仍然希望看到统一交易契约

结论：

• 当前商品中心采用 “主干交易契约 + 垂直资源域” 的组织模型
• 商品中心维护的是全平台统一的 售卖货架与交易契约
• 真正复杂的资源真相下沉到各自垂直业务域治理，例如：
  • 酒店域维护 hotel_id / room_type / date -> price & availability
  • 电影域维护 cinema_id / show_id / seat -> inventory
  • 充值域维护 carrier / amount / channel -> price & capability

进一步的落地结论是：

1. 商品中心维护的是 Meta Item，而不是全量物理资源
   • 不为每一家物理酒店、每一个场次、每一个充值通道都创建重商品对象
   • 商品中心只维护少量可交易的“品类元商品”或统一货架契约
2. 交易链路通过动态参数携带具体资源上下文
   • 例如 hotel_id、room_type、biz_date、phone_number、carrier_code
   • 商品中心在交易时负责返回统一交易契约，不负责承载所有资源细节
3. 订单快照通过运行时多态合成
   • 订单侧拿到 item_id + ext_params
   • 商品中心以主干 Item 为底稿，按需向垂直资源域取当时真相，合成 item_snapshot
   • 这样既保证统一交易格式，又不把高频资源变化硬塞进商品中心主表
4. 生命周期采用双层状态防御
   • 商品中心状态只表达“平台是否允许售卖这一类商品契约”
   • 具体某家酒店、某个场次、某条充值通道是否可售，由垂直资源域继续判定

一句话说：

商品中心写的核心，不是去记录这个世界上所有物理资源的变化，而是去维护全平台统一的交易货架与售卖契约。让商品中心保持轻量，让资源复杂度回归垂直业务域，通过运行时快照在交易发生的瞬间完成真相交汇。

4.3 商品中心核心模型

商品中心当前只显式建模这 3 个核心对象：

• Draft
• Item
• Item Snapshot

4.3.1 Draft

• 待审核、待发布、可编辑
• 是供给平台与商品中心交互时最主要的写入对象
• 同一时刻只保留一个当前 Draft
• 提交审核后会被锁定，直到审核结论返回或人工撤回

4.3.2 Item

• 只表达正式可交易商品
• 承担对外读模型主权
• 不能混入草稿、审核中的流程态
• publish_version 当前作为 Item 的内联字段存在，是并发保护与正式版本推进的锚点

4.3.3 Item Snapshot

• 是订单解释事实
• 订单只信交易时绑定的 item_snapshot
• 不依赖“当前线上商品是否又被改过”

4.3.4 对象关系与核心字段

4.4 核心功能：创建商品草稿与审核态锁定

4.4.1 场景画像

这个场景解决的是：

• 供给平台提交一份标准化商品 DTO
• 商品中心不直接生成正式 Item
• 而是先创建一个可审核、可锁定、可追溯的 Draft

核心目标是：

• 创建草稿不等于正式发布
• 审核对象必须稳定
• 供给侧拿到的是 draft_id，不是正式商品主权

4.4.2 完整时序图

sequenceDiagram
    participant S as "供给平台"
    participant P as "商品中心"
    participant D as "Draft"

    S->>P: CreateDraft(标准 DTO, operation_id)
    P->>D: 创建 Draft
    D-->>P: 返回 draft_id
    P-->>S: 返回 draft_id / current_publish_version

    S->>P: SubmitDraft(draft_id)
    P->>D: 读取当前 Draft
    P->>D: status = QC_PENDING，锁定草稿
    P-->>S: 返回 draft_id / base_publish_version

4.4.3 关键技术点

4.5 核心功能：商品编辑、版本冲突与增量保护

4.5.1 场景画像

这个场景解决的是：

• 运营或供给侧基于当前正式版本做编辑
• 多人并发或异步补偿重放时，如何防止旧版本覆盖新版本
• 最终的 Diff、字段主导权和版本裁决应该由谁负责

4.5.2 完整时序图

sequenceDiagram
    participant S as "供给平台"
    participant P as "商品中心"
    participant I as "正式 Item"
    participant D as "Draft"

    S->>P: 加载当前正式对象(item_id)
    P->>I: 读取正式 Item / publish_version
    I-->>P: 返回 current_publish_version / 当前 DTO
    P-->>S: 返回 current_publish_version

    S->>P: UpdateDraft(draft_id, base_publish_version, 增量字段)
    P->>I: 校验当前 publish_version
    alt 版本不匹配
        P-->>S: VERSION_CONFLICT
    else 版本匹配
        P->>D: 更新 Draft
        P-->>S: DraftUpdated
    end

4.5.3 关键技术点

商品中心对“是否跳过本次更新”的判断，建议遵循下面这条心智：

1. 先读取当前正式 Item DTO 与 publish_version
2. 依据字段主导权矩阵裁掉供应商无权覆盖的字段
3. 对保留下来的有效字段做轻量内存 Diff
4. 如果有效字段完全一致，则直接判定为 NO_OP 并返回成功
5. 如果存在有效变化，再进入正式的版本校验和本地事务 merge

在 Go 里的典型实现，可以是这种“强类型快路径 + 版本锚点兜底”的组合：

type SPUTruth struct {
    Title     string   `json:"title"`
    BrandID   int64    `json:"brand_id"`
    Price     int64    `json:"price"`
    ImageURLs []string `json:"image_urls"`
}

func ShouldSkipUpdate(currentLive SPUTruth, incomingReq SPUTruth) bool {
    if currentLive.Title == incomingReq.Title &&
        currentLive.BrandID == incomingReq.BrandID &&
        currentLive.Price == incomingReq.Price &&
        sliceEqual(currentLive.ImageURLs, incomingReq.ImageURLs) {
        return true
    }
    return false
}

这里的 ShouldSkipUpdate() 只负责过滤掉 完全没有有效变化 的请求；真正的并发保护仍然依赖：

• base_publish_version
• product_item_tab.publish_version
• merge 时的乐观锁或版本校验

4.6 核心功能：审核输入、发布编排与本地 Merge

4.6.1 场景画像

这个场景解决的是：

• 商品中心如何接收 QC verdict
• verdict 与当前锁定草稿不一致时如何阻断
• 发布时如何在本地事务中完成 merge，而不是依赖跨服务长事务

4.6.2 完整时序图

sequenceDiagram
    participant Q as "QC 平台"
    participant P as "商品中心"
    participant D as "Draft"
    participant I as "正式 Item"
    participant SNAP as "Item Snapshot"
    participant O as "Outbox"

    Q->>P: ReceiveQcVerdict(draft_id, publish_version, result)
    P->>D: 校验 draft_id / status=QC_PENDING / base_publish_version
    alt verdict 不匹配 / 版本已演进
        P-->>Q: REJECT_STALE_VERDICT
    else PASS
        P->>I: 本地事务 merge 到正式 Item
        P->>I: publish_version = publish_version + 1
        P->>SNAP: 写 item_snapshot
        P->>O: 写发布事件
        P->>D: status = PUBLISHED
        P-->>Q: PUBLISH_ACCEPTED
    end

4.6.3 关键技术点

4.7 核心功能：商品编辑历史与审计轨迹

4.7.1 场景画像

这个场景解决的是：

• 运营、商家或系统在商品中心里到底改过什么
• 谁在什么时间点修改了哪些字段
• 一次修改到底属于草稿编辑、审核流转，还是正式发布
• 当线上商品出现争议时，如何回放“这版商品是怎么变成现在这样”的路径

这里的目标不是做一份通用日志，而是沉淀一条 商品中心可解释的编辑历史链路：

• 面向运营，能回答“谁改了什么”
• 面向排障，能回答“这版数据从哪来”
• 面向审计，能回答“这是 Draft、QC，还是 Publish 阶段的动作”
• 面向前台展示，能直接渲染出“字段从旧值变成新值”的结构化 Diff，而不是只给一条“某某改了商品”的流水文本

因此这里不能只做一张泛化操作日志，而是要先把历史分成 3 类，再决定每类记录什么：

1. Draft 历史

   • 表示草稿编辑阶段的字段变更
   • 重点回答：谁改了哪些字段
   • 典型动作：EDIT_DRAFT / WITHDRAW / RESUBMIT

2. QC 历史

   • 表示审核流转阶段的状态动作
   • 重点回答：谁在什么时候提交审核、驳回、重新提交、通过
   • 典型动作：SUBMIT_FOR_QC / REJECT / APPROVE / ABORT
   • QC 历史更偏“流程动作”，不一定总有字段级 Diff

3. Publish 历史

   • 表示锁定草稿最终如何进入正式态
   • 重点回答：哪一个 Draft 在什么时候发布成了哪一个正式版本
   • 典型动作：PUBLISH / ROLLBACK / OFFLINE
   • Publish 历史要和 publish_version、item_snapshot 绑定，保证正式版本可解释

这意味着商品中心保存的不应该只是“操作事件”，而应该是 对象级 / 字段级的变更差异。例如：

• 商品标题：iPhone 15 基础版 -> iPhone 15 降价促销版
• 商品底价：5999.00 -> 5499.00
• 发货时效：48h -> 24h

而不是仅仅记录：

• 张三在 2026-06-05 12:00 修改了商品

4.7.2 完整时序图

sequenceDiagram
    participant S as "供给平台 / 运营后台"
    participant P as "商品中心"
    participant D as "Draft"
    participant H as "Edit History"
    participant I as "正式 Item"

    S->>P: UpdateDraft(draft_id, base_publish_version, 增量字段, operator)
    P->>D: 读取当前 Draft
    P->>D: 更新 Draft
    P->>H: 记录字段级 Diff(change_payload, operator)
    P-->>S: DraftUpdated

    S->>P: SubmitDraft(draft_id, operator)
    P->>H: 记录 QC 历史(action=SUBMIT_FOR_QC)
    P-->>S: Submitted

    S->>P: Publish(draft_id, operator)
    P->>I: merge 到正式 Item
    P->>H: 对比旧 Item 与锁定 Draft，记录 Publish 历史(action=PUBLISH, publish_version, change_payload)
    P-->>S: PublishAccepted

4.7.3 关键技术点

建议的 change_payload 形态示例：

[
  {
    "field": "title",
    "label": "商品标题",
    "old": "iPhone 15 基础版",
    "new": "iPhone 15 降价促销版"
  },
  {
    "field": "price",
    "label": "商品底价(元)",
    "old": 5999.00,
    "new": 5499.00
  },
  {
    "field": "detail_html",
    "label": "商品详情",
    "changed": true
  }
]

在 Go 里的实现建议是：

1. 默认方案：轻量反射 Diff

   • 基于 reflect 遍历 DTO 字段
   • 结合结构体标签控制：
     • 字段展示名
     • 是否参与 Diff
   • 输出统一的 change_payload=[{field, label, old, new}]

2. 复杂嵌套补充：go-cmp / JSON Patch

   • 如果对象层级非常深，可以引入：
     • github.com/google/go-cmp/cmp
     • github.com/evanphx/json-patch
   • 但它们更适合作为辅助工具，不建议直接把原始文本 Diff 结果塞进主日志表

3. 高性能优化：代码生成

   • 若事务内反射成本过高，可以为核心 DTO 生成专用 Diff 方法
   • 例如：

func (old *ProductDTO) CustomDiff(new *ProductDTO) []ChangeLogDetail

• 用原生 if old.Price != new.Price 这类静态比较替换反射
• 适合高频发布、深对象和严格事务延迟场景

一句话说：

Go 里的 DiffUtil 不追求“最炫的通用框架”，而追求“结构化输出、事务内足够轻、重 Diff 可以延后”。

前台展示心智应该是：

• 操作人：张三
• 时间：2026-06-05 12:00:00
• 生效版本：V3
• 变更内容：
  • 修改了商品标题：iPhone 15 基础版 -> iPhone 15 降价促销版
  • 修改了商品底价：5999.00 -> 5499.00
  • 修改了商品详情：点击查看 Diff

4.8 发布后读模型与外部投影

商品中心正式发布后，要统一驱动：

• 搜索索引刷新
• 缓存刷新
• 计价上下文更新
• 营销圈品
• 数据平台投影
• 订单快照读取

这里的边界要明确：

• 商品中心负责发布正式事件
• 外部系统自己消费并刷新投影
• 订单只认 item_snapshot，不认“此刻线上商品”

4.9 异常处理与版本恢复

这一节要回答的不是“如何永不失败”，而是失败后如何解释、恢复、阻断污染。

商品中心写链路的终极心智是：

在商品中心主域里，永远不要试图去“修改”或“倒退”历史。所有的修改、发布、下架、封禁，甚至是回滚，在底层的物理世界上，都是一次携带着最新版本锚点的正向时间演进（+1）。只要把住这条铁律，商品中心的写模型就立于不败之地。

目标是：

• 阻断旧版本回滚污染
• 保证正式态始终可解释
• 补偿不跨越商品中心真相边界

4.10 商品中心表清单与使用场景

4.10.1 每个表的使用场景

如果后续第 5 章把 QC 平台表完全独立出来，则：

• product_qc_review 在第 3 章只作为交互对象提及
• 不应列为商品中心权威表

4.10.2 每个表的 Schema

product_draft_tab

product_item_tab

item_snapshot

product_edit_history_log

5. 详细设计 - 库存中心

5.1 库存中心定位与职责边界

库存中心管理的不是一个简单数字，而是平台对用户的一种 可承诺供给能力。

它负责：

• 库存对象与范围建模
• 可售量、预占、确认、释放
• 券码池、锁库存、账本
• 对账、修复、热视图恢复

它不负责：

• B 端商品编辑
• 商品审核流程
• 正式商品资产主权

5.2 决策点

5.2.1 决策点 1：库存中心是否只管理一个数字

结论：

• 库存中心管理的不是一个单纯数字，而是用户可被平台承诺的供给能力
• 余额只是当前视图，真正的库存事实还包括预占、账本和资源实例

5.2.2 决策点 2：库存是否统一按 SKU 管，还是按 inventory_key 管

结论：

• 库存不天然只按 SKU 管，而是按“承诺范围”管
• 推荐以 inventory_key 统一表达范围，典型维度包括：
  • 商品 / SKU
  • 门店
  • 日期
  • 渠道
  • 批次
  • 券码池 / 资源池

5.2.3 决策点 3：热路径 Redis 与 MySQL 权威账本的职责分离

结论：

• Redis 负责 C 端热路径裁决
• MySQL 负责 库存中心唯一的权威余额与权威账本
• Redis 只是 可丢弃、可重建、可漂移的热视图
• 所有落库解释、冲正修复、全量恢复都必须以 MySQL 为准，绝不能反向用 Redis 改写账本

一句话说：

Redis 解决“现在能不能卖得出去”，MySQL 解决“库存最终为什么会变成这样”。

5.2.4 决策点 4：数量库存、系统发券库存、外部死码库存能否共用一套事实模型

结论：

• QTY / SYS_GEN / EXT_POOL 可以共用命令层表达
• 但事实模型必须分流
• 尤其 EXT_POOL 必须走资源型库存，不允许退化为一串数字

5.2.5 决策点 5：库存调整应该传绝对值还是 Delta

结论：

• B 端看到的是绝对值
• 但库存中心只接受带 base_inventory_version 的 Delta 命令
• 库存事实的版本裁决必须由库存中心完成

5.2.6 决策点 6：资源型库存（券码、座位、房态）是否应该折算成普通数字库存

结论：

• 资源型库存不能折算成普通数字库存
• 数字只是影子，权威真相仍在资源实例表

5.2.7 决策点 7：供应商实时库存是否直接写库存中心正式账本

结论：

• 供应商实时库存不应直接盲写库存中心正式账本
• 应先经过资源域或库存命令层受控进入

5.2.8 决策点 8：库存编辑中的 +10 / -10 增量操作，应该直接改余额，还是走原子增量 + 账本幂等

结论：

• 库存中心必须把 +10 / -10 这类操作建模成 增量命令
• 余额表更新必须采用数据库原子更新，不能先查再算再覆盖
• 同一个事务内必须同时完成两件事：
  • 向 inventory_ledger 写入一条增量流水
  • 对 inventory_balance 执行原子增量更新
• 所有增量命令都必须携带全局唯一的 request_no / operation_id
• inventory_ledger 必须对该请求号建立唯一索引，用于拦截网络超时重试带来的重复加减

典型落地形态是：

UPDATE inventory_balance
SET usable_qty = usable_qty + :delta,
    updated_at = NOW()
WHERE inventory_key = :inventory_key
  AND (usable_qty + :delta) >= 0;

以及：

UNIQUE KEY uk_request_no (request_no)

一句话说：

库存中心里真正的真相不是“当前余额是多少”，而是“每一次 +10 / -10 到底是谁、在什么时候、基于什么请求号改进去的”。余额只是聚合结果，账本和请求号幂等才是准确性的底座。

5.2.9 决策点 9：库存追加或调整的审批流，应该放在库存中心还是供给平台

结论：

• 审批流属于 业务流程主权，应放在供给平台
• 库存中心只负责 事实执行主权
• 供给平台应负责：
  • 发起库存调整申请
  • 审批流转
  • 操作理由、附件、OA 单号、权限校验
  • 审批通过后的正式命令编排
• 库存中心只接收“已经获批、允许执行”的命令，例如：
  • AdjustInventory(delta, request_no, approval_no)
  • ImportCodeBatch(task_id, approval_no)
  • GenerateCodeBatch(task_id, request_no, approval_no)

这样可以形成稳定分层：

• 供给平台负责“谁申请、谁审批、为什么改”
• 库存中心负责“到底加了多少、减了多少、写进了哪条账本、当前余额变成多少”

一句话说：

审批属于流程治理，应该放在供给平台；库存中心只做库存事实的受控执行，不演化成审批工作流平台。

5.3 库存中心核心模型

库存中心建议显式建模以下对象：

这里的模型边界要钉死：

• inventory_balance 不是审计真相，只是当前聚合视图
• inventory_ledger 才是库存中心的最终解释事实
• inventory_code_pool_xx 是资源型库存的主权表，不应被压扁成数字字段

5.4 核心功能：B 端库存初始化与库存类型路由

5.4.1 场景画像

这个场景解决的是：

• 创建商品时如何同步初始化库存
• 已有商品如何单独创建库存
• inventory_mode = QTY / SYS_GEN / EXT_POOL 如何路由到不同初始化路径

库存中心面对的不是一个“永远只有数字”的世界，而是三类差异很大的初始化入口：

• QTY：直接初始化数量库存
• SYS_GEN：库存中心负责闭环生成券码，供给平台只负责发起生成任务和跟踪进度
• EXT_POOL：初始化动作不接受手填数量，而是等待唯一资源导入来生成数量影子

5.4.2 场景一：QTY 数量库存初始化时序图

sequenceDiagram
    participant S as "供给平台"
    participant IC as "库存中心"
    participant CFG as "inventory_config"
    participant BAL as "inventory_balance"

    S->>IC: CreateInventory(item_id, inventory_mode=QTY, init_quantity)
    IC->>CFG: 写库存配置
    IC->>BAL: 初始化 available_qty
    IC-->>S: INIT_SUCCESS

5.4.3 场景二：EXT_POOL 外部死码导入初始化时序图

sequenceDiagram
    autonumber
    actor O as "运营 / 供应商"
    participant S as "供给平台"
    participant IC as "库存中心"
    participant CFG as "inventory_config / inventory_balance"
    participant POOL as "inventory_code_pool_xx"
    participant MQ as "Message Queue / Outbox Event"

    O->>S: 上传券码 Excel
    S->>S: 生成 task_id，状态=PROCESSING
    S->>IC: CreateInventory(item_id, inventory_mode=EXT_POOL, task_id)
    IC->>CFG: 创建配置，init_status=PENDING_IMPORT, usable_qty=0
    IC-->>S: INIT_PENDING

    loop 分批导码
        S->>S: 清洗本批券码，去除文件内重复
        S->>IC: ImportCodeBatch(item_id, codes_chunk[], task_id)
        IC->>POOL: INSERT IGNORE 批量落库
        IC-->>S: 返回本批成功数
    end

    S->>IC: CompleteImport(item_id, task_id)
    IC->>POOL: COUNT 有效券码真相
    IC->>CFG: usable_qty = real_count, init_status = READY
    IC->>MQ: 事务内写完成事件
    MQ-->>S: INVENTORY_IMPORT_COMPLETED(task_id, real_count)
    S->>S: CAS 更新任务状态为 SUCCESS
    S-->>O: 前台进度条完成 / 通知完成

5.4.4 场景三：SYS_GEN 系统活码生成与回调时序图

sequenceDiagram
    autonumber
    actor O as "运营 / 供应商"
    participant S as "供给平台"
    participant IC as "库存中心"
    participant W as "Inventory Gen Worker"
    participant CFG as "inventory_config / inventory_balance"
    participant POOL as "inventory_code_pool_xx"
    participant MQ as "Message Queue / Outbox Event"

    O->>S: 发布商品并要求系统生成 target_qty 个券码
    S->>S: 初始化生成任务 task_id，状态=PROCESSING
    S->>IC: CreateInventory(item_id, inventory_mode=SYS_GEN, target_qty, task_id)
    IC->>CFG: 创建配置，init_status=PENDING_GEN, usable_qty=0
    IC-->>S: TASK_ACCEPTED
    S-->>O: 商品已发布，后台造码中

    IC->>W: 异步触发内部造码 Worker
    loop 直到物理有效码数达到 target_qty
        W->>W: 生成一批带盐随机券码
        W->>POOL: INSERT IGNORE 批量落库
        W->>POOL: COUNT 当前真实有效码数
    end

    W->>CFG: usable_qty = target_qty, init_status = READY
    W->>MQ: 事务内写 INVENTORY_GEN_COMPLETED 事件
    MQ-->>S: INVENTORY_GEN_COMPLETED(task_id, real_count)
    S->>S: CAS 更新任务状态为 SUCCESS
    S-->>O: 进度条拉满 / 通知完成

5.4.5 场景四：QTY 数量库存追加时序图

sequenceDiagram
    autonumber
    actor O as "运营 / 供给平台"
    participant IC as "库存中心"
    participant BAL as "inventory_balance"
    participant LED as "inventory_ledger"

    O->>IC: AdjustInventory(item_id, delta=+10, request_no, base_inventory_version)
    IC->>LED: 事务内插入 +10 流水（request_no 唯一）
    IC->>BAL: usable_qty = usable_qty + 10（原子更新）
    IC-->>O: ADJUST_SUCCESS

5.4.6 场景五：EXT_POOL 外部死码追加时序图

sequenceDiagram
    autonumber
    actor O as "运营 / 供应商"
    participant S as "供给平台"
    participant IC as "库存中心"
    participant POOL as "inventory_code_pool_xx"
    participant LED as "inventory_ledger"
    participant BAL as "inventory_balance"

    O->>S: 上传包含 10 个新券码的 Excel
    S->>S: 创建追加任务 task_id，状态=PROCESSING

    loop 分批导码
        S->>IC: ImportCodeBatch(item_id, codes_chunk[], task_id)
        IC->>POOL: INSERT IGNORE 批量落库
        IC-->>S: 返回本批真实成功行数
    end

    S->>IC: CompleteImport(item_id, task_id)
    IC->>POOL: COUNT 本次任务真实新增有效码数
    IC->>LED: 事务内写 delta_qty = +real_count 流水
    IC->>BAL: usable_qty = usable_qty + real_count
    IC-->>S: IMPORT_SUCCESS(real_count)
    S-->>O: 提示“上传 10 个码，成功导入 9 个，库存追加 9”

5.4.7 场景六：SYS_GEN 系统活码追加时序图

sequenceDiagram
    autonumber
    actor O as "运营 / 供应商"
    participant S as "供给平台"
    participant IC as "库存中心"
    participant W as "Inventory Gen Worker"
    participant POOL as "inventory_code_pool_xx"
    participant LED as "inventory_ledger"
    participant BAL as "inventory_balance"
    participant MQ as "Message Queue / Outbox Event"

    O->>S: 输入“追加 10 个系统券码”
    S->>S: 创建追加任务 task_id，状态=PROCESSING
    S->>IC: GenerateCodeBatch(item_id, target_qty=10, task_id, request_no)
    IC-->>S: TASK_ACCEPTED

    IC->>W: 异步触发内部造码 Worker
    loop 直到真实新增 10 个有效码
        W->>W: 生成一批带盐随机券码
        W->>POOL: INSERT IGNORE 批量落库
        W->>POOL: COUNT 本次任务真实新增有效码数
    end

    W->>LED: 事务内写 delta_qty = +10 流水
    W->>BAL: usable_qty = usable_qty + 10
    W->>MQ: 事务内写 INVENTORY_GEN_COMPLETED 事件
    MQ-->>S: INVENTORY_GEN_COMPLETED(task_id, real_count=10)
    S->>S: CAS 更新任务状态为 SUCCESS
    S-->>O: 追加完成，库存已增加 10

5.4.8 关键技术点

5.5 核心功能：C 端预占、确认、释放与订单链路联动

5.5.1 场景画像

这个场景解决的是：

• 下单时如何先预占库存，避免超卖
• 支付失败、超时取消或风控拦截时如何归还库存
• 支付成功后如何确认扣减，并把库存状态推进到可履约
• 履约发货或到店核销时，如何继续沿着订单生命周期推进资源状态

库存中心不能只表达“当前剩余多少”，还必须承接 C 端交易过程态：

• AVAILABLE
• RESERVED
• CONFIRMED
• RELEASED
• LOCKED
• CONSUMED

5.5.2 场景一：下单预占库存时序图

sequenceDiagram
    participant O as "订单中心"
    participant IC as "库存中心"
    participant BAL as "inventory_balance"
    participant RES as "inventory_reservation"
    participant LED as "inventory_ledger"

    O->>IC: ReserveInventory(order_id, inventory_key, qty)
    IC->>BAL: 扣减可售 / 写预占视图
    IC->>RES: 写 reservation
    IC->>LED: 写预占账本
    IC-->>O: RESERVED

5.5.3 场景二：支付失败、超时取消与风控拦截后的库存归还时序图

sequenceDiagram
    participant O as "订单中心"
    participant IC as "库存中心"
    participant BAL as "inventory_balance"
    participant RES as "inventory_reservation"
    participant LED as "inventory_ledger"

    O->>IC: ReleaseInventory(order_id)
    IC->>BAL: 回补可售
    IC->>RES: 标记 RELEASED
    IC->>LED: 写释放账本
    IC-->>O: RELEASED

5.5.4 场景三：支付成功后的库存确认时序图

sequenceDiagram
    participant O as "订单中心"
    participant IC as "库存中心"
    participant RES as "inventory_reservation"
    participant LED as "inventory_ledger"

    O->>IC: ConfirmInventory(order_id)
    IC->>RES: 标记 CONFIRMED
    IC->>LED: 写确认账本
    IC-->>O: CONFIRMED

5.5.5 场景四：实物发货或履约完成后的库存消费时序图

sequenceDiagram
    participant O as "订单中心 / 履约中心"
    participant IC as "库存中心"
    participant RES as "inventory_reservation"
    participant LED as "inventory_ledger"

    O->>IC: ConsumeInventory(order_id)
    IC->>RES: 标记 CONSUMED
    IC->>LED: 写消费 / 履约完成账本
    IC-->>O: CONSUMED

5.5.6 场景五：到店核销或券码核销时序图

sequenceDiagram
    participant O as "订单中心 / 核销中心"
    participant IC as "库存中心"
    participant RES as "inventory_reservation"
    participant LED as "inventory_ledger"

    O->>IC: VerifyAndConsume(order_id, code)
    IC->>RES: 标记 CONSUMED
    IC->>LED: 写核销消费账本
    IC-->>O: VERIFIED_AND_CONSUMED

5.5.7 关键技术点

5.6 核心功能：券码池与唯一资源型库存

5.6.1 场景画像

这个场景解决的是：

• 导码
• 生码
• 发码
• 回补
• 坏码、失效码和重复发放的治理

券码库存不是数字库存的一个附属字段，而是一套独立的一码一实例模型。

5.6.2 完整时序图

sequenceDiagram
    participant S as "供给平台 / 运营后台"
    participant IC as "库存中心"
    participant POOL as "inventory_code_pool_xx"
    participant BAL as "inventory_balance"
    participant O as "订单中心"

    S->>IC: ImportCodeBatch(item_id, code_rows)
    IC->>POOL: 批量写入唯一资源
    IC->>BAL: 根据有效资源行数回写聚合数量
    IC-->>S: IMPORT_SUCCESS

    O->>IC: ReserveInventory(order_id, item_id, qty=1)
    IC->>POOL: 锁定一张可用券码
    IC->>BAL: 扣减聚合数量
    IC-->>O: RESERVED(code_id)

    alt 支付成功 / 发码成功
        O->>IC: ConfirmInventory(order_id)
        IC->>POOL: 标记券码 CONFIRMED / USED
    else 取消 / 失败
        O->>IC: ReleaseInventory(order_id)
        IC->>POOL: 券码回补为可用
        IC->>BAL: 回补聚合数量
    end

5.6.3 关键技术点

5.7 热视图、账本与高并发治理

5.7.1 场景画像

这一节只回答库存中心在 C 端高并发交易热路径 下的热点治理问题，不重复 5.8 的异常矩阵。

当面对大促、爆款秒杀、热门房态和热门场次时，如果让每一次 ReserveInventory 都直接穿透到 MySQL：

• inventory_balance 单行会形成排他锁串行排队
• 订单链路 RT 会被瞬间拉高
• 热点对象会把交易数据库拖入雪崩

因此库存中心在 C 端场景下采用的是：

• Redis 做热路径状态裁决
• Redis 内同步推进逻辑位点 seq
• MQ 做削峰与异步落库
• MySQL 做权威账本、权威余额和已处理水位线
• 定时对账 + 动态冲正 + 极端容灾重建 做最终一致修复

这里的核心口径是：

• Redis 是“出纳快照”，负责快速判断能不能卖，并在热路径内推进逻辑时钟
• MySQL 是“会计真相”，负责解释每一笔库存到底怎么变来的，并记录当前已经处理到哪个逻辑位点
• 视图允许短期漂移，但账本必须最终正确
• 只有当 Redis 与 MySQL 的逻辑位点完全对齐时，系统才进入“时空静默”窗口，允许执行安全冲正

5.7.2 C 端高并发预占与异步落库时序图

sequenceDiagram
    autonumber
    participant O as "订单中心"
    participant IC as "库存中心"
    participant R as "Redis 热路径裁决"
    participant MQ as "Message Queue / Outbox Event"
    participant W as "Inventory Persist Worker"
    participant BAL as "MySQL inventory_balance"
    participant LED as "MySQL inventory_ledger"

    O->>IC: ReserveInventory(order_id, inventory_key, qty)
    IC->>R: 执行 Lua 原子预占脚本(扣减 + 推进 seq)

    alt Redis 裁决库存不足
        R-->>IC: return INSUFFICIENT
        IC-->>O: RESERVE_REJECTED
    else Redis 裁决预占成功
        R-->>IC: return RESERVED, seq=N
        IC->>R: HSET reserve_cache:{order_id}
        IC->>MQ: 写 INVENTORY_RESERVE_EVT(order_id, inventory_key, qty, seq=N)
        IC-->>O: RESERVED

        MQ->>W: 异步消费预占事件
        W->>BAL: SELECT last_processed_seq FOR UPDATE
        alt seq 已落后于 DB 水位线
            W->>LED: 写存证流水（order_id 作为 request_no）
            W-->>MQ: ACK_STALE_EVENT
        else seq 正常推进
            W->>LED: 写 RESERVE 流水（order_id 作为 request_no, seq=N）
            W->>BAL: 事务内更新权威余额 + last_processed_seq=N
            W-->>MQ: ACK
        end
    end

5.7.3 Redis Lua 热路径原语

为了防止“先 GET 再 SET”带来的竞态条件，Redis 热路径必须走 Lua 原子脚本，而不能在应用层拆成多次命令。

典型预占脚本思路如下：

local key_avail = "inventory:available_qty:" .. KEYS[1]
local key_reserve = "inventory:reserved_qty:" .. KEYS[1]
local key_seq = "inventory:seq:" .. KEYS[1]
local qty = tonumber(ARGV[1])

local current_avail = redis.call('GET', key_avail)
if not current_avail or tonumber(current_avail) < qty then
    return {-1, 0}
end

redis.call('DECRBY', key_avail, qty)
redis.call('INCRBY', key_reserve, qty)
local new_seq = redis.call('INCR', key_seq)
return {1, new_seq}

这条脚本的边界要讲清楚：

• Lua 负责热路径原子裁决与逻辑位点推进
• Lua 不负责生成最终解释账本
• MQ 消息必须携带这个 seq，把 Redis 热路径和 MySQL 慢路径绑在同一条逻辑时间线上
• 真正的库存真相仍以后续落到 inventory_ledger + inventory_balance 的结果为准

5.7.4 热视图漂移、单向对账与动态冲正

由于 Redis 天然存在：

• AOF / RDB 异步落盘窗口
• 主从切换丢失瞬时状态
• 异步消息重放乱序
• B 端直接改账本而热视图未同步

所以 Redis 与 MySQL 之间出现短期漂移是系统设计内允许发生的事情。

库存中心不追求“Redis 和 DB 永远瞬时强一致”，而是通过 逻辑水位线 + 定时准实时对账 + 动态冲正修复 来收敛。

对账 Worker 至少会取到 4 个关键值：

• GET inventory:seq:{key}
• GET inventory:available_qty:{key}
• inventory_balance.last_processed_seq
• inventory_balance.available_qty

随后分 3 种情况处理：

1. redis_seq > mysql_seq
   • 说明高并发热路径仍在向前推进，异步落库仍在追赶
   • 正常情况下直接 IN_FLIGHT_SKIP
   • 若 MySQL 很久没有推进，说明异步消费可能卡死，应报警而不是贸然修复
2. redis_seq == mysql_seq
   • 说明 Redis 与 MySQL 在逻辑时间线上已经完全对齐，进入可安全对账窗口
   • 如果数量也一致，则系统健康
   • 如果数量不一致，则以 MySQL 为准计算 drift_delta，对 Redis 执行 增量冲正
3. redis_seq < mysql_seq
   • 正常生产环境中几乎不应该出现
   • 一旦出现，通常意味着 Redis 重启、热视图丢失或极端故障恢复后的基准回退
   • 此时应触发 Redis 冷启动重建，而不是普通增量修补

这里的原则必须锁死：

• 绝不能用 Redis 覆盖 MySQL
• 只能用 MySQL 的权威真相去重建 Redis 热视图
• 所谓“对账”不是 Redis 和 MySQL 平等互相纠偏，而是 Redis 向 MySQL 单向收敛
• 冲正也不应直接 SET 绝对值覆盖，而应优先采用基于差值的增量修补；只有 Redis 崩溃冷启动这类极端场景，才允许全量基准重建

5.7.5 Redis 全量故障下的受控慢路径降级

如果 Redis 分片集群整体故障，库存中心仍要尽量保证核心交易不中断，只是吞吐和 RT 降级。

推荐策略是：

1. 配置中心下发降级开关，热路径切到 “DB 受控慢路径”
2. 网关或订单入口对热点商品执行限流、排队和熔断
3. 预占直接走 MySQL 条件更新，例如：

UPDATE inventory_balance
SET available_qty = available_qty - :qty,
    reserved_qty = reserved_qty + :qty,
    last_processed_seq = last_processed_seq + 1
WHERE inventory_key = :inventory_key
  AND available_qty >= :qty;

4. Redis 恢复后，由对账任务按 MySQL 当前余额和预占量全量重刷热视图
5. 确认重建完成后，再切回 Redis 热路径

这条链路的核心目标不是“性能不受影响”，而是：

• 在 Redis 不可用时，交易链路仍能 带降级地活下去
• 在 Redis 恢复后，热视图能 重新从权威账本拉平
• 在 Redis 位点归零时，库存中心还能基于 last_processed_seq 把逻辑时钟顶回正确基准

对于 Redis 崩溃且位点丢失的极端情况，推荐专门走“冷启动重建”：

1. 读取 inventory_balance.available_qty / reserved_qty / last_processed_seq
2. 使用高危受控脚本重置：
   • SET inventory:available_qty:{key} = mysql_available
   • SET inventory:reserved_qty:{key} = mysql_reserved
   • SET inventory:seq:{key} = mysql_last_processed_seq
3. 记录 inventory_repair_task
4. 放开流量并恢复 Redis 热路径

当某个商品首次跌到 0 时，还应向应用层广播“售罄信号”，让 App 节点在本地短期缓存 is_sold_out:{inventory_key}，把后续洪峰尽量挡在 Redis 之前。

5.7.6 关键技术点

5.8 异常处理与对账修复

库存中心不是“永远不出错”，而是出错后必须可解释、可回放、可对账、可修复。

5.9 库存中心表清单与使用场景

5.9.1 每个表的使用场景

5.9.2 每个表的 Schema

inventory_config

inventory_balance

inventory_reservation

inventory_ledger

inventory_code_pool_xx

inventory_reconcile_task

inventory_repair_task

6. 详细设计 - QC 和治理

6.1 QC 平台定位与职责边界

这里的 QC 指的是 商品侧 / 供给侧审核中心，不指库存质检。

QC 平台负责：

• 机审核
• 人工审核
• 审核编排
• 审核审计
• 审核治理看板

QC 平台不负责：

• 正式商品主权
• 商品本地 merge
• 库存事实

6.2 为什么 QC 要独立于供给平台和商品中心

QC 独立出来的价值在于：

• 审核规则演进更快
• 人工审核工作台可以独立建设
• 策略编排和审核审计不污染供给平台
• 风控和合规逻辑不耦合进商品中心事务链路

6.3 机审核设计

机审核不是“写几条正则就完事”，而是一套把高频可规则化风险前移拦截的执行系统。它的目标不是完全替代人工，而是把明显正确和明显错误的对象尽量自动化，让人工审核只处理真正需要判断的灰区。

机审核典型覆盖以下风险面：

• 敏感词与违规描述
• 类目规则与属性约束
• 主图 / 素材基础规则
• 资质与证照完整性
• 价格异常与履约承诺异常
• 交易契约缺失，例如退款规则、预约规则、有效期缺失

从执行链路看，推荐拆成 4 层：

1. 基础结构校验层：字段必填、格式、范围、枚举、跨字段依赖。
2. 类目规则层：类目模板、属性完整性、品类专属约束。
3. 风控合规模型层：敏感词、资质、素材风险评分。
4. 发布风险裁决层：根据规则命中结果给出 PASS / REJECT / MANUAL_REVIEW。

推荐使用“规则 + 策略路由 + 风险分值”三段式模型，而不是把所有结论硬编码在一个 if/else 里：

推荐把机审核结果统一收敛为三类：

• 自动通过：低风险且规则全部满足，可以直接进入发布或自动准入队列。
• 自动驳回：存在确定性违规，必须回退给供给侧修改。
• 转人工审核：规则只能判断出“有风险但不够确定”，需要人工裁决。

这样设计的关键价值在于：规则系统只负责“发现风险”，真正的发布裁决仍然有一层独立的 verdict 模型，便于后续插入人工审核、风控加签、特殊品类白名单等流程。

6.4 人工审核设计

人工审核要解决的不是“再看一眼”，而是把不确定性风险转化为可运营、可追责、可度量的工作流系统。很多团队的问题不在规则本身，而在审核池失控、抢单无序、标准不统一和积压无法消化。

一个可用的人工审核系统，至少要解决 5 件事：

1. 审核对象怎么进池。
2. 审核任务怎么分发给合适的人。
3. 审核员看到的是哪一版冻结快照。
4. 驳回和复审怎么闭环。
5. 审核结果怎么沉淀成规则优化输入。

推荐的人审工作台能力包括：

在执行模型上，推荐把人工审核任务显式建模成“工作项”，而不是直接在商品草稿上打审核字段。因为审核任务天然带有：

• 队列属性
• 领取属性
• SLA 属性
• 升级属性
• 审计属性

如果这些能力都塞回商品表或供给任务表里，后面只会越来越难治理。

6.5 审核路由与分级策略

审核绝不能一刀切。真正的平台会根据 变更内容、品类风险、来源可信度、历史行为、发布时间窗口 做分层路由。

推荐至少从 4 个维度做路由：

一个常见但错误的做法是：只要改了商品就一律进人工审核。这样短期看“稳”，长期一定拖垮平台效率。

更合理的路由策略通常是：

1. 低风险字段小改动：例如副标题、运营标签、小幅价格修正，可走自动准入。
2. 中风险结构性变更：例如履约规则、退款条件、素材变化，先过机审，再视命中情况转人工。
3. 高风险关键变更：例如类目迁移、资质变更、上下架、供应商关键属性替换，强制进入人工审核。
4. 平台强制裁决类事件：例如风控下架、法务拦截，可绕过普通审核队列，走高优先级裁决链路。

审核分级的本质不是“谁都审一下”，而是用最小的人力成本把最大的风险拦下来。

6.6 审核对象与冻结快照

这一节是 QC 设计最关键的地方。QC 审的一定是 冻结快照，不能审供给侧正在被编辑的动态草稿，也不能直接审线上正式商品表。

原因很简单：

• 如果审核员看到的是动态草稿，那么在审核过程中，运营或商家继续编辑，就会出现“审的是 A，发的是 B”。
• 如果审核员直接看正式商品表，就会把未生效变更和已上线资产混在一起，失去版本边界。

推荐采用以下对象关系：

推荐链路如下：

1. 供给平台或商品中心维护可编辑 Draft。
2. 用户点击“提交审核”后，商品中心生成新的 publish_version。
3. 系统把对应字段冻结成送审 snapshot。
4. QC 只对 snapshot 做机审和人审。
5. QC 返回 verdict 时，必须带回 draft_id + publish_version + snapshot_id 或等价冻结标识。
6. 商品中心只允许对“当前待发布版本”执行 merge 和正式发布。

flowchart LR
    A["可编辑 Draft"] --> B["提交审核"]
    B --> C["冻结 Snapshot"]
    C --> D["机审核 / 人工审核"]
    D --> E["返回 verdict + publish_version"]
    E --> F["商品中心本地 merge 发布"]

这套模型的核心价值不是“多建一个快照表”，而是把审核对象从“会继续变化的编辑态”切成“可追责的冻结态”。这样后面无论是申诉、回放、抽检还是事故复盘，都知道当时到底审了什么。

6.7 审核 verdict 与商品中心交互

QC 平台回传的应该是 轻量 verdict，而不是整份商品 DTO。QC 的职责是裁决，不是替商品中心做主数据写入。

推荐回传结构如下：

{
  "draft_id": 555,
  "publish_version": 102,
  "snapshot_id": 900123,
  "result": "PASS",
  "review_level": "AUTO",
  "reason_codes": ["TITLE_OK", "PRICE_OK"]
}

商品中心收到 verdict 后，再在本地事务里完成 merge 和正式发布。这里必须坚持两个边界：

1. QC 不直写正式商品。
2. 商品中心不信任“没有版本号的审核结果”。

推荐交互规则如下：

也就是说，QC 更像法官，商品中心更像执行机关。法官只给结论，不帮你改资产；执行机关拿到结论后，在自己的事务边界里决定状态怎么推进、正式态怎么落库、事件怎么广播。

6.8 QC 治理闭环

审核系统真正难的地方，往往不是单次“过还是不过”，而是长周期治理闭环。一个成熟的 QC 平台必须让错误能修、积压能控、规则能进化、事故能回放。

推荐重点治理以下 8 类问题：

治理闭环里最有价值的两个运营指标通常是：

1. 审核吞吐指标：自动通过率、人工命中率、平均审核时长、SLA 超时率。
2. 审核质量指标：误判率、申诉成功率、上线后追罚率、漏拦截事故数。

只有把这两类指标一起看，平台才能知道自己是在“提效”，还是只是在“把风险放过去”。

6.9 QC 表清单与使用场景

推荐把 QC 平台最少拆成以下几类表：