主题
架构演进：每拆开一个边界，都要重新回答谁负责提交和恢复
任务管理系统刚上线时，一个进程里创建任务、更新状态、发送通知，几个人就能维护。后来报表拖慢了任务列表，通知故障影响创建，两个团队又总要等对方一起发布。此时需要演进，但这三种问题未必需要同一种拆法。
本文沿着这个假设系统说明单体、垂直拆分、服务化和事件驱动怎样解决不同问题，再分别解释 CQRS、事件溯源与读写模型分离。主线是数据责任：谁有权修改一项事实，什么必须一起提交，其他人看到旧状态时怎么办，以及迁移失败后怎样退回。
先分清这些词在拆什么
| 概念 | 拆开的对象 | 可以仍在同一进程／数据库吗 | 新增的主要成本 |
|---|---|---|---|
| 模块化单体 | 代码与业务职责 | 可以，通常同一部署单元 | 需要持续维护模块边界 |
| 垂直拆分 | 业务能力，如任务、成员、通知 | 可以先做逻辑拆分，再独立部署 | 跨能力依赖与数据归属要明确 |
| 服务化 | 部署、调用与数据所有权边界 | 可以暂用同一数据库服务器，但不应随意互写表 | 网络失败、契约、运维、跨服务一致性 |
| 事件驱动 | 状态变化与后续处理的时间耦合 | 可以同进程，也可以跨服务 | 延迟、重复、乱序、事件版本、排障 |
| CQRS | 命令模型与查询模型 | 可以共用同一数据库 | 两套模型的维护和一致性解释 |
| 事件溯源 | 权威事实的保存方式 | 可以单体，也可以服务化 | 历史事件兼容、回放、快照和隐私管理 |
| 数据库读写分离 | 写节点与只读副本的执行路径 | 数据模型可以完全相同 | 复制滞后、读写路由、切换与读一致性 |
这不是必须按顺序走完的成熟度阶梯。一个业务边界清楚、流量可控的模块化单体可以长期工作；一个独立报表模型也不要求把所有业务都改成事件溯源。选型要对应已观察到的问题。
单体的第一步：让模块先能独立理解
单体指作为一个主要单元构建和部署，不等于所有代码都放在同一个包。任务、成员、通知可以各自有应用接口和内部数据访问。创建任务只依赖成员模块提供的“是否有权限”能力，不直接到处读取成员模块的内部表字段。
正在绘制图示…
查看图源代码
flowchart LR
H["HTTP 入口"] --> T["任务模块：创建、改状态"]
H --> M["成员模块：权限、团队"]
T -->|"通过明确接口"| M
T --> N["通知模块接口"]
subgraph D["一个数据库事务环境"]
TD[("任务表与创建结果")]
MD[("成员表")]
OD[("outbox")]
end
T --> TD
M --> MD
T --> OD
OD -.-> N
模块边界要落实到四件事：谁拥有表和业务规则；哪些接口允许其他模块调用；哪些数据结构不能向外泄漏；哪些操作必须在同一事务内完成。代码包分好了，却允许每个模块直接更新其他模块的表，之后仍很难拆开。
好处是本地调用便宜、事务直接、调试和发布路径短。代价是资源争抢和统一发布仍在，一个模块的内存失控可能影响整个进程。可以先把慢报表放到后台任务、限制连接池、优化查询，未必立刻拆成微服务。Fowler 的 Monolith First讨论了先发现稳定边界再拆分的理由，也明确其经验性；它不是“所有系统必须从单体开始”的定律。
判断是否值得拆，可先收集具体证据：一个业务模块是否需要不同扩容节奏，是否被独立团队长期维护，发布互相等待是否确实增加交付时间，以及隔离故障的价值能否覆盖多服务运维成本。“代码行数多”本身不足以回答这些问题。
垂直拆分：围绕业务能力，别只把层变成服务
垂直拆分可以先得到任务、成员和通知三个业务模块，也可以将某个模块独立部署。它与“前端服务、逻辑服务、数据库服务”这种按技术层切开不同：若每次业务变化仍要同时改三层服务，部署数增加了，业务耦合却没减少。
这个案例先拆通知比较合理：通知允许延迟，外部渠道故障不应阻塞创建，吞吐和重试模式也不同。任务与成员权限可能仍保留较紧密的同步关系，因为创建时需要回答“这个人现在能否操作”。边界不是由目录名决定，而是由事务与业务时效决定。
| 拆分候选 | 希望获得什么 | 拆之前必须补齐什么 |
|---|---|---|
| 通知 | 渠道故障隔离、独立 worker 扩容 | 持久通知意图、幂等消费、超时和重试上限 |
| 报表 | 慢查询不占核心请求资源 | 可接受的数据滞后、可重建的数据来源 |
| 成员与授权 | 明确团队所有权、独立权限演进 | 权限缓存失效、撤权生效时间、依赖不可用策略 |
| 任务核心 | 独立部署与数据生命周期 | 请求幂等、并发版本、迁移时单写者 |
拆分后的“数据库归某服务所有”通常意味着其他服务不能绕过接口直接改它的业务数据，不一定要求立刻使用不同数据库产品或物理服务器。先有明确所有权，再根据隔离、资源和维护需求决定物理部署。共享数据库中跨服务 JOIN 很方便，却会让内部 schema 变成隐式公共契约，需要明确它是临时迁移方案还是长期约束。
服务化：进程边界变成失败边界
一个本地函数调用要么返回、要么进程失败；远程调用还会遇到超时但对方已完成、请求没到、响应丢了、连接断开等情况。拆服务时必须补上 deadline、取消、幂等业务键、有限重试、并发上限、身份与授权、链路关联和错误分类。
如果创建任务先同步调用通知，再提交数据库，通知可能已经发出但任务回滚；如果先提交任务再发通知，进程可能在两步之间崩溃。把这两行移到不同服务，并不会自动出现一个跨服务事务。
正在绘制图示…
查看图源代码
sequenceDiagram
participant A as 任务服务
participant D as 任务数据库
participant R as outbox 转发器
participant N as 通知服务
A->>D: 单事务写任务与 TaskCreated 事件意图
D-->>A: 提交成功
R->>D: 读取待发送事件
R->>N: 发送 event_id
N->>N: 去重与创建通知放在本地事务
N-->>R: 已接受
R->>D: 记录投递进度
转发器在发送后、记录进度前崩溃，会重发；所以通知服务按 event_id 去重。outbox 解决本地提交与外部投递之间的丢失窗口，不承诺外部副作用绝对只发生一次。原模式说明见 Transactional Outbox。
在拆分前，至少画出一个失败时序，而不只画成功调用链。尤其要问：下游已成功但本地不知道时，查询哪个结果？是否允许重复？哪条状态能证明工作完成？这些答案比服务数量更影响可靠性。
事件驱动：发布已经发生的事实
命令表达希望发生的动作，如 CompleteTask；事件表达已经发生的事实，如 TaskCompleted。命令可能被拒绝，事件应能对应一个已提交的业务变化。事件里只写“数据变了，来查一下”会减少载荷，却使消费者强依赖生产者当时可用；携带需要的事实能降低查询耦合，但增加复制的数据、隐私和版本维护成本。
一个最小事件信封可以包含 event_id、event_type、schema_version、aggregate_id、aggregate_version、occurred_at、payload，以及追踪所需的 correlation_id。时间戳用于展示和诊断；不能只按墙上时间推断业务先后。对于同一个任务，递增版本更适合发现重复和缺口；多个任务之间通常不需要全局顺序。
| 契约问题 | 本案例的选择 | 为什么要明确 |
|---|---|---|
| 重复投递 | 允许；消费者持久去重 | 发送成功与 ACK 丢失无法只靠发送方区分 |
| 单任务顺序 | 按 task_id 分区，并检查 aggregate_version | 换分区或并发消费时可能破坏顺序 |
| 未知版本 | 暂停或隔离并告警 | 静默跳过会让投影永久缺数据 |
| 消费进度 | 与本地投影更新原子提交，或用可重放的等价协议 | 先写数据后记进度会重复，反过来会漏处理 |
| 积压 | 有限保留、按优先级隔离，监控最老事件年龄 | 队列长度不能直接表示业务延迟 |
| schema 兼容 | 新旧消费者共存期内保持兼容，先升级能读新格式的消费者 | 否则一次发布会阻断所有下游 |
这里的“原子提交”依赖具体数据库与消息系统。事件日志与投影库是两个系统时，不能靠一句“事务”把它们包起来。可以把已处理事件 ID 与投影同库提交，消息确认放到提交之后，允许重复读取再去重；也可以使用目标产品明确支持的事务协议。Kafka 的事务与 exactly-once 能力有适用边界，见 Kafka 4.1 消息语义。
事件驱动让通知不阻塞创建，却会增加看见全流程的难度。排障时需要从请求 ID 找到任务版本，再找到 event_id、消费结果和投递状态。把所有日志打得更多，不能替代这些关联键和状态查询入口。
CQRS：写入规则与查询形状可以分别设计
CQRS 是 Command Query Responsibility Segregation：命令负责改变状态，查询负责返回信息。任务的写模型可能围绕“只有未完成任务能改标题”“修改时版本必须匹配”；看板的读模型则需要负责人姓名、团队名称、状态统计和最近活动。让一套结构同时满足两者，可能导致写规则散落或查询昂贵。
最小 CQRS 可以只是同一进程中的命令接口与查询接口，仍读写同一个数据库。再往前一步，命令更新规范化表，查询读取单独维护的看板投影。只有需要独立性能、可用性或存储形态时，才进一步拆开数据库或服务。定义和复杂度边界见 Fowler CQRS 与 Azure CQRS 模式。
正在绘制图示…
查看图源代码
flowchart LR
C["命令：完成任务，expected_version=2"] --> W["写模型：验证状态和版本"]
W --> A[("权威当前状态或事件流")]
A -->|"已提交变化，可能延迟"| P["投影器"]
P --> R[("看板读模型")]
Q["查询：团队看板"] --> R
W -->|"写结果及版本 3"| U["客户端"]
R -->|"结果及已处理版本"| U
图中的权威数据可以是普通当前状态表，也可以是事件流，所以 CQRS 不要求事件溯源。反过来，采用事件溯源也不强制每一种查询都使用独立服务。
投影异步更新时，写成功后立即读取看板可能还看不到变化。可以直接把命令返回的新状态显示在客户端；可以让后续查询带 min_version，并在有限等待内等投影追上；也可以把需要写后读的请求路由到权威模型。等待必须有超时和回退，不能把“最终一致”变成无限转圈。
本机模型的权威事件流已经写到版本 2，但读投影只处理了版本 1：
text
write_version=2 read_version=1 title=初稿旧页面再带 expected_version=1 提交修改，返回：
text
stale command: version conflict版本检查保护写状态，读模型滞后仍要通过接口解释。它们解决的是两个不同问题。
事件溯源：把变化序列作为权威事实
普通持久化常把任务当前标题覆盖为新标题。事件溯源保存“创建任务”“改名”“完成”等已提交事件，再由事件推导当前状态。例如 t1 的事件序列是 Created(v1) → Renamed(v2) → Completed(v3)。当前标题与完成标志可以重建，过去发生过什么也有来源。
这与普通审计日志不同：如果删掉事件就无法可靠重建权威状态，那么事件是主要存储；如果当前表才是权威，旁边日志可能漏记也不影响主流程，就不能只因“有事件表”称为事件溯源。事件驱动关注交互方式，事件溯源关注事实存储方式，两者可以单独采用。Fowler Event Sourcing和Azure Event Sourcing分别讨论了这一模型及其使用条件。
| 能得到什么 | 必须付出的成本 |
|---|---|
| 保留业务变化历史，可构建新投影 | 历史事件 schema 长期可读，修错流程更复杂 |
| 能重放到某个版本检查状态 | 回放耗时与快照管理；外部依赖不能按“今天的值”乱重算历史 |
| 同一事实派生不同读模型 | 多投影的一致性、滞后、重建资源与版本管理 |
| 通过追加修正事件保留原因 | 删除个人数据、保留政策和不可变日志之间需要专门设计 |
事件应该包含重放所需的确定事实。比如事件记录的是当时采用的业务结果与必要参数，而不是在回放时重新访问外部接口获取今日结果。事件处理函数也不要在内部读随机数、当前时钟或可变远端状态来决定历史状态，否则相同事件可能重建出不同结果。
写入时对每个聚合检查 expected_version，再追加下一事件。本文把一个任务当聚合，即一次规则检查与原子修改的边界。模型的关键部分是：
go
if expected != len(hs) {
return Event{}, errors.New("version conflict")
}
e.Seq = len(hs) + 1完整实现还先 upcast，再用 reducer 验证合法状态，最后才追加。测试让 10 个 goroutine 争着从版本 1 改名，只有一个成功。它使用内存 mutex 表示原子边界；没有磁盘提交、跨进程锁或集群复制。生产还需独立 command_id 幂等结果：提交后响应丢失时，仅靠 expected_version 冲突，客户端仍不知道是不是自己的命令已经成功。模型没有实现这项生产能力。
投影如何处理重复、缺口和重建
投影把一个事件转换为可查询状态。本文要求同一任务逐版本应用，重复 event_id 且内容相同则直接成功；同一个 ID 内容不同则报错。遇到版本缺口，先暂停该任务的推进并查缺失事件，不能直接跳到最新版本。
go
if old, ok := p.Seen[e.ID]; ok {
if old != e { return errors.New("event ID payload conflict") }
return nil
}把版本 2 应用两遍，输出仍是：
text
after replay: version=2 title=审校把版本 3 直接发给空投影，输出：
text
gap: sequence gap这里的版本是聚合内部序号。消息系统的 partition offset 则是某个分区中的读取位置，不是任务版本；一个分区可能混着许多任务。还可能有快照版本与投影代码版本，不应塞进同一个 version 字段里。
正在绘制图示…
查看图源代码
flowchart LR
E[("历史事件 + 持续新增事件")]
E --> O["旧投影，继续服务"]
E --> N["新投影：独立进度从头重建"]
N --> C["追平水位、校验样本与总量"]
O --> R["查询路由"]
C -->|"满足切换条件后"| R
R --> U["用户查询"]
重建时不要清空正在服务的读表后再慢慢回放。建立新投影，记下回放范围与水位，继续消费新增事件，核对结果，再切查询路由。保留旧投影一段明确的回退窗口。注意：这里的路由回退容易，写模型和事件语义若已改变，回退到旧写程序未必安全。
快照保存某个聚合在版本 N 的状态，加载时只回放 N 之后的事件。快照是性能优化，要包含版本、schema 和必要校验信息；若只是可丢弃缓存，就不能在历史事件已删后仍声称能从头重建。保留多少期事件、多久能重建、重建会占用多少资源，需要作为运行要求测量。
还有一个容易遗漏的事务：更新投影与保存已处理标记必须一致。受控反例先把计数加一，在记录 seen 前模拟崩溃，再重投，输出：
text
broken split commit count=2示例测试明确验证错误发生，不靠真正杀进程。正确方向是在同一数据库事务内写投影与去重／进度记录，或设计等价的幂等赋值协议。内存示例只能表达这个边界，不能验证某个数据库的崩溃恢复。
schema 演进：老事件不会随新代码一起消失
假设旧事件 Created 的 schema 1 字段叫 Name，新版 schema 2 叫 Title。不能简单让新代码只认 Title，否则重建旧数据时标题会变空。模型保留旧事件，读取时 upcast 成统一内部结构：
go
switch e.Schema {
case 1:
c.Title = e.Name
case 2:
c.Title = e.Title
default:
return Canonical{}, fmt.Errorf("unsupported schema %d", e.Schema)
}这是字段重命名的确定映射，不是任意补全缺失事实。旧事件从未记录的数据不能凭空恢复；若必须推导，保存推导依据、版本与不确定性，必要时走一次显式迁移，而不是悄悄给所有历史事件填今天的默认值。
| 变更 | 常见处理 | 检查点 |
|---|---|---|
| 新增可选字段 | 老消费者忽略，新消费者接受缺省 | 缺省是否真的保持原语义 |
| 字段改名或类型改变 | 版本化事件，读取时 upcast | 所有旧版本有测试夹具 |
| 业务含义改变 | 引入新事件类型或明确新版本 | 不能只改名字继续复用旧含义 |
| 修复历史错误 | 追加更正／撤销事实，必要时受控修复 | 保留审计链；所有投影如何反映更正 |
| 个人数据删除 | 减少事件中的个人信息，使用受控引用等方案 | 保留法律与产品要求需专门评估，不能承诺删一行就完成 |
重建结果和现有投影一致，才说明当前 reducer 对这组事件稳定。它不证明旧、新 schema 在所有历史输入上兼容，所以要保留真实的历史样本、未知版本负例和业务边界测试。
重放时，别把通知再发一遍
如果处理 TaskCompleted 时既更新看板又发邮件，重新建看板就可能重发所有历史邮件。把纯状态变换与外部副作用分开：reducer 只算状态；需要发通知的消费者有自己的幂等记录、处理进度和投递策略。重建读模型时根本不走发送通道。
模型先执行一次独立通知消费者，再重建投影：
text
rebuild equal=true done=true notifications=1
retry notifications=0notifyCompleted 只是向内存 map 记账并计数，没有真实发信。生产中“发送成功”和“记录发送成功”之间仍有故障窗口，能否避免重复取决于接收端幂等和状态查询能力。不能从这两行输出推导邮件、支付等外部动作恰好执行一次。
读写分离：同一数据模型，也可能读到旧数据
数据库主节点写、只读副本读，常用于分担查询。这与 CQRS 的模型职责分离不是同一件事：主从可以仍用同一张 tasks 表；CQRS 也可以不使用副本。只读副本的复制可能滞后，新任务写入主库后立即读副本可能暂时不存在。具体数据库还可能因回放与长查询冲突取消查询，见 PostgreSQL 18 Hot Standby。
| 读取要求 | 可选策略 | 代价 |
|---|---|---|
| 允许几秒滞后的报表 | 只读副本或异步投影 | 标注刷新时间并监控实际滞后 |
| 本人刚写后必须看到 | 返回写结果，或短期路由主库 | 主库读负载增加；固定时间窗不一定覆盖最坏滞后 |
| 需要至少读到某次提交 | 使用系统支持的提交水位／版本令牌并有限等待 | 客户端与路由更复杂；不支持时不能假装有保证 |
| 权限撤销、库存资格判断 | 在明确权威边界做判定 | 不能因读副本更空闲就读旧授权或库存 |
复制副本也不是备份：错误删除可以被复制过去。是否能回到错误发生前，要看独立备份、日志保留、恢复程序与演练结果。
跨服务流程：补偿是新业务动作
服务拆开后，原来一笔事务可能变成多个本地事务。例如培训系统先预留名额，再创建报名；创建失败，需要释放预留。Saga 通过一系列本地事务和相应补偿组织流程，可以由协调者推进，也可以由事件触发；定义见 Saga 模式。
| 步骤 | 本地提交的事实 | 后续失败时可能的补偿 |
|---|---|---|
| 预留名额 | reservation=reserved | 条件释放该预留，避免重复返还 |
| 创建报名 | enrollment=confirmed | 追加取消报名状态 |
| 发送确认邮件 | 提供方接受了消息 | 无法撤回已读邮件，必要时发送更正说明 |
补偿不像数据库回滚那样让世界恢复成完全没发生过。名额释放前可能有其他人操作，邮件可能已经被读到，补偿自己也会超时或失败。每一步与补偿都需要幂等键、可查询状态、重试期限和人工异常队列；并且要确定哪些步骤可以重试完成、哪些必须补偿。参见 Compensating Transaction。
Saga 不提供与单库事务相同的隔离。中间状态可能被其他流程看到；若业务要求整个过程绝对不可见，可能需要保留更大的事务边界、使用预留状态或重新设计交互，而不是给每个步骤补一条事件就认为问题解决。
迁移：先证明新路径正确，再交出写入权
渐进替换可以在统一入口后按业务或租户切换旧、新实现，相关思路见 Strangler Fig。关键是写入所有权。让两个系统长期独立写同一份业务事实，然后靠定时任务“最终同步”，会使冲突与回退难以解释。
一次有界迁移可以这样执行：
- 在旧系统内建立模块接口和唯一写入口，先消除绕过模块的写操作。
- 把新 schema 或新服务部署成兼容旧数据的状态；用历史夹具、契约测试和失败场景验证。
- 从一致的基线回填，再持续消费增量，检查遗漏、重复、分区进度和权限数据。
- 对只读请求做影子比较，核对结果与延迟；影子路径不执行邮件、扣款等副作用。
- 按明确租户或分区切换写入所有者。暂停旧写或使用受控路由／版本屏障，保证同一对象不会双写。
- 验证新写、新读和恢复路径，保留可操作的回退方案；确认没有旧消费者后，再移除旧字段和旧路径。
schema 的 expand-contract 也遵循这个顺序：先添加兼容结构，让新旧版本都能运行；迁移并校验；最后才删除旧结构。数据库加列、回填和索引构建是否锁表、占多少空间，必须按具体数据库版本测试。代码回滚不等于数据迁移自动回滚，删掉旧列后再运行旧二进制通常就来不及了。
如果拆分后跨服务调用过多、每次发布仍须全体协调、故障定位变慢，而隔离收益很小，可以把边界重新合并。演进的验收指标应是交付、故障隔离、容量与恢复能力是否改善，不能只数服务或消息主题增加了多少。
依据与验证边界
- Fowler Monolith First、Strangler Fig：模块边界与渐进替换的原作者讨论。
- Fowler CQRS、Azure CQRS：命令、查询模型与数据同步。
- Fowler Event Sourcing、Azure Event Sourcing：权威事件、重建、快照及采用成本。
- Transactional Outbox、Saga、Compensating Transaction：本地事务、消息和跨服务恢复边界。
- Kafka 4.1 Design、PostgreSQL 18 Hot Standby：投递语义与读副本限制。本文没有搭建这两个产品做实验。
- 示例在 2026-10-13、Go 1.27.2、Linux/amd64 实际运行。完整输出如正文；普通测试、race 和 vet 通过。包含旧／新 schema、重复与同 ID 异参、版本缺口、未知 schema、非法事件无写入、10 个并发命令只有一个成功、无通知副作用的重建及错误分步提交反例。
- 本模型只验证内存状态机。事件持久化、数据库投影事务、消息系统崩溃恢复、生产回填和跨服务补偿没有实测。迁移步骤是设计建议，需要在目标系统上形成有停止条件和回退条件的实施计划。
完整代码与测试
把下列文件放在同一目录。程序无外部依赖、网络服务和文件持久化；所有任务与事件都是固定教学数据。Store 的写入用 mutex 保护；Projection 与通知消费者按顺序运行，没有宣称可被多个 goroutine 随意并发调用。
sh
go run .
go test -count=1 -v ./...
go test -race -count=1 ./...
go vet ./...go.mod
go.mod
module example.com/evolution
go 1.27.0main.go
代码块 · 187 行 · 点击收起 / 展开
go
package main
import (
"errors"
"fmt"
"slices"
"sync"
)
type Event struct {
ID, Task, Kind string
Seq, Schema int
Name, Title string
}
type Canonical struct {
ID, Task, Kind, Title string
Seq int
}
func upcast(e Event) (Canonical, error) {
c := Canonical{ID: e.ID, Task: e.Task, Kind: e.Kind, Seq: e.Seq}
switch e.Schema {
case 1:
c.Title = e.Name
case 2:
c.Title = e.Title
default:
return Canonical{}, fmt.Errorf("unsupported schema %d", e.Schema)
}
return c, nil
}
type State struct {
Title string
Done bool
Version int
}
func reduce(s State, e Canonical) (State, error) {
if e.Seq != s.Version+1 {
return s, errors.New("sequence gap")
}
switch e.Kind {
case "Created":
if s.Version != 0 || e.Title == "" {
return s, errors.New("invalid create")
}
s.Title = e.Title
case "Renamed":
if s.Version == 0 || s.Done || e.Title == "" {
return s, errors.New("invalid rename")
}
s.Title = e.Title
case "Completed":
if s.Version == 0 || s.Done {
return s, errors.New("invalid complete")
}
s.Done = true
default:
return s, errors.New("unknown event kind")
}
s.Version = e.Seq
return s, nil
}
// Atomic in-memory append and expected-version check. No disk persistence.
type Store struct {
mu sync.Mutex
History map[string][]Event
}
func (s *Store) Append(expected int, e Event) (Event, error) {
s.mu.Lock()
defer s.mu.Unlock()
hs := s.History[e.Task]
if expected != len(hs) {
return Event{}, errors.New("version conflict")
}
e.Seq = len(hs) + 1
e.ID = fmt.Sprintf("%s:%d", e.Task, e.Seq)
state := State{}
for _, old := range hs {
c, err := upcast(old)
if err != nil {
return Event{}, err
}
state, err = reduce(state, c)
if err != nil {
return Event{}, err
}
}
c, err := upcast(e)
if err != nil {
return Event{}, err
}
if _, err = reduce(state, c); err != nil {
return Event{}, err
}
s.History[e.Task] = append(hs, e)
return e, nil
}
func (s *Store) Events(task string) []Event {
s.mu.Lock()
defer s.mu.Unlock()
return slices.Clone(s.History[task])
}
// This projection is called serially. Production needs transactional update+offset.
type Projection struct {
Rows map[string]State
Seen map[string]Event
}
func NewProjection() *Projection {
return &Projection{Rows: map[string]State{}, Seen: map[string]Event{}}
}
func (p *Projection) Apply(e Event) error {
if old, ok := p.Seen[e.ID]; ok {
if old != e {
return errors.New("event ID payload conflict")
}
return nil
}
c, err := upcast(e)
if err != nil {
return err
}
next, err := reduce(p.Rows[e.Task], c)
if err != nil {
return err
}
// Logical atomic boundary; no external calls in this reducer.
p.Rows[e.Task] = next
p.Seen[e.ID] = e
return nil
}
func rebuild(es []Event) (*Projection, error) {
p := NewProjection()
for _, e := range es {
if err := p.Apply(e); err != nil {
return nil, err
}
}
return p, nil
}
// An independent, explicitly invoked effect consumer. Rebuild never invokes it.
func notifyCompleted(es []Event, delivered map[string]bool) int {
n := 0
for _, e := range es {
if e.Kind == "Completed" && !delivered[e.ID] {
delivered[e.ID] = true
n++
}
}
return n
}
func must(err error) {
if err != nil {
panic(err)
}
}
func main() {
s := &Store{History: map[string][]Event{}}
p := NewProjection()
e1, err := s.Append(0, Event{Task: "t1", Kind: "Created", Schema: 1, Name: "初稿"})
must(err)
must(p.Apply(e1))
e2, err := s.Append(1, Event{Task: "t1", Kind: "Renamed", Schema: 2, Title: "审校"})
must(err)
fmt.Printf("write_version=%d read_version=%d title=%s\n", e2.Seq, p.Rows["t1"].Version, p.Rows["t1"].Title)
_, err = s.Append(1, Event{Task: "t1", Kind: "Renamed", Schema: 2, Title: "旧页面"})
fmt.Println("stale command:", err)
must(p.Apply(e2))
must(p.Apply(e2))
fmt.Printf("after replay: version=%d title=%s\n", p.Rows["t1"].Version, p.Rows["t1"].Title)
e3, err := s.Append(2, Event{Task: "t1", Kind: "Completed", Schema: 2})
must(err)
fmt.Println("gap:", NewProjection().Apply(e3))
must(p.Apply(e3))
delivered := map[string]bool{}
sent := notifyCompleted(s.Events("t1"), delivered)
rebuilt, err := rebuild(s.Events("t1"))
must(err)
fmt.Printf("rebuild equal=%v done=%v notifications=%d\n", rebuilt.Rows["t1"] == p.Rows["t1"], rebuilt.Rows["t1"].Done, sent)
fmt.Printf("retry notifications=%d\n", notifyCompleted(s.Events("t1"), delivered))
}main_test.go
代码块 · 97 行 · 点击收起 / 展开
go
package main
import (
"sync"
"sync/atomic"
"testing"
)
func TestEvolution(t *testing.T) {
s := &Store{History: map[string][]Event{}}
e1, err := s.Append(0, Event{Task: "t", Kind: "Created", Schema: 1, Name: "old"})
if err != nil {
t.Fatal(err)
}
e2, err := s.Append(1, Event{Task: "t", Kind: "Renamed", Schema: 2, Title: "new"})
if err != nil {
t.Fatal(err)
}
p := NewProjection()
if err = p.Apply(e2); err == nil || len(p.Rows) != 0 {
t.Fatal("accepted gap")
}
for _, e := range []Event{e1, e1, e2, e2} {
if err = p.Apply(e); err != nil {
t.Fatal(err)
}
}
if p.Rows["t"] != (State{Title: "new", Version: 2}) {
t.Fatal(p.Rows)
}
bad := e2
bad.Title = "tampered"
if err = p.Apply(bad); err == nil {
t.Fatal("accepted payload conflict")
}
bad = e2
bad.ID = "bad"
bad.Schema = 99
if err = p.Apply(bad); err == nil {
t.Fatal("accepted unknown schema")
}
rebuilt, err := rebuild(s.Events("t"))
if err != nil || rebuilt.Rows["t"] != p.Rows["t"] {
t.Fatal(err)
}
if _, err = s.Append(2, Event{Task: "t", Kind: "Unknown", Schema: 2}); err == nil || len(s.Events("t")) != 2 {
t.Fatal("invalid append mutated history")
}
}
func TestOptimisticConcurrency(t *testing.T) {
s := &Store{History: map[string][]Event{}}
_, err := s.Append(0, Event{Task: "t", Kind: "Created", Schema: 2, Title: "first"})
if err != nil {
t.Fatal(err)
}
var success atomic.Int32
var wg sync.WaitGroup
for range 10 {
wg.Go(func() {
if _, err := s.Append(1, Event{Task: "t", Kind: "Renamed", Schema: 2, Title: "second"}); err == nil {
success.Add(1)
}
})
}
wg.Wait()
if success.Load() != 1 || len(s.Events("t")) != 2 {
t.Fatal(success.Load())
}
}
func TestRebuildDoesNotNotify(t *testing.T) {
es := []Event{{ID: "t:1", Task: "t", Kind: "Created", Seq: 1, Schema: 1, Name: "old"}, {ID: "t:2", Task: "t", Kind: "Completed", Seq: 2, Schema: 2}}
delivered := map[string]bool{}
if notifyCompleted(es, delivered) != 1 {
t.Fatal("missing live effect")
}
if _, err := rebuild(es); err != nil {
t.Fatal(err)
}
if notifyCompleted(es, delivered) != 0 {
t.Fatal("duplicate effect")
}
}
// Counterexample: applying a projection before remembering progress duplicates a delta.
func TestSplitCommitCounterexample(t *testing.T) {
count := 0
seen := false
count++ // crash before seen=true
if !seen {
count++
seen = true
}
if count != 2 || !seen {
t.Fatal(count, seen)
}
t.Logf("broken split commit count=%d", count)
}