10.3 事件溯源与审计
玩家申诉:「我昨天明明有 3 件装备,今天只剩 1 件,是不是你们吞了?」客服打开后台,看到的是玩家当前的背包——1 件。当前状态里没有历史,谁也说不清昨天那 2 件去哪了。
10.2 的对账能发现「两边不一致」,但它也回答不了「为什么不一致」。要回答「为什么」,系统必须记得每一步。这就是事件溯源(event sourcing)要解决的事:不存「结果」,存「发生过的事」,结果由事件重放算出来。
这一节给资产系统装上事件流、快照与重放,并写一个可跑的审计查询。所有代码在 internal/saga/audit/ 包,只用标准库。
事件流、快照与重放
事件溯源有三块积木。
**事件流(event stream)**是 append-only 的:每发生一次资产变动,就追加一条不可变的事件,记下「谁、什么类型、哪个物品、变了多少、什么时间、关联哪个单号」。事件一旦写入就不改不删——这是它和「状态表」最根本的区别:状态表只记得现在,事件流记得全部过去。
**快照(snapshot)**是性能优化。事件攒到几十万条,每次重放都要从头算太慢,所以定期把「重放到此刻的状态」存下来。查历史时从最近一个快照往后重放即可,不必从第一条开始。
重放(replay)是核心操作:把事件按序折叠成一个状态。state = fold(events),且这个折叠必须是纯函数——同样的输入永远得到同样的输出。重放是纯的,才有后面两个能力:恢复(进程崩了,重放事件重建内存状态)和审计(把某段时间的事件挑出来,算出资产怎么变的)。
三者关系可以用一句话概括:事件流是真相,快照是缓存,重放是求值。任何时刻,「当前状态」都应该等于「重放全部事件」的结果;如果不相等,说明有事件丢了或被偷偷改了——这本身就是一条最强的自洽校验。
可跑的事件存储与重放
下面这段是 L1,go test 可以直接跑。它定义了事件、状态、存储,以及两个关键方法:Replay(折叠事件成状态)和 Explain(回答「某段时间资产为什么变」)。
// file: internal/saga/audit/audit.go
// tier: L1
// verify: go test ./internal/saga/audit/
package audit
// Event is one immutable line of a player's asset history. The log is
// append-only: nothing is ever edited or deleted, so history can always be
// replayed to prove the current state.
type Event struct {
Seq uint64 // strictly increasing, assigned on append
PlayerID uint64
Type string // "grant" / "spend" / "cross_settle" / "rollback"
ItemID uint32
Delta int64 // positive to grant, negative to spend
At int64 // server-stamped unix milliseconds
Ref string // correlation id, e.g. a battle or order id
}
// State is the asset snapshot produced by replaying a prefix of the log.
type State struct {
Seq uint64
At int64
Assets map[uint32]int64
}
// Store is an in-memory append-only event log with periodic snapshots.
type Store struct {
events []Event
snaps map[uint64]State // player -> latest snapshot
next uint64
}
// NewStore returns an empty log.
func NewStore() *Store {
return &Store{snaps: map[uint64]State{}, next: 1}
}
// Append stamps and stores one event, returning the assigned sequence number.
func (s *Store) Append(e Event) uint64 {
e.Seq = s.next
s.next++
s.events = append(s.events, e)
return e.Seq
}
// Events returns every event for a player with Seq in [from, to] inclusive.
func (s *Store) Events(pid, from, to uint64) []Event {
var out []Event
for _, e := range s.events {
if e.PlayerID == pid && e.Seq >= from && e.Seq <= to {
out = append(out, e)
}
}
return out
}
// Snapshot records the state at the current head. Replay must always reproduce
// this exact value, which is the whole point of an audit log.
func (s *Store) Snapshot(pid uint64, at int64) State {
st := Replay(s.Events(pid, 0, ^uint64(0)))
st.At = at
s.snaps[pid] = st
return st
}
// Replay folds events into asset state. It is pure: the same events always
// produce the same State, so it can be re-run on demand or after a restore.
func Replay(events []Event) State {
st := State{Assets: map[uint32]int64{}}
for _, e := range events {
st.Assets[e.ItemID] += e.Delta
if st.Assets[e.ItemID] == 0 {
delete(st.Assets, e.ItemID)
}
st.Seq = e.Seq
}
return st
}
// Explanation answers "why did this player's assets change". It returns the
// before/after state of the window plus the events that caused the change, so a
// support engineer can read the story instead of diffing two tables.
type Explanation struct {
PlayerID uint64
From int64
To int64
Before State
After State
Events []Event
}
// Explain replays the log up to the window start, then within the window, and
// returns both the boundary states and the in-window events.
func (s *Store) Explain(pid uint64, from, to int64) Explanation {
all := s.Events(pid, 0, ^uint64(0))
ex := Explanation{PlayerID: pid, From: from, To: to}
ex.Before = State{Assets: map[uint32]int64{}}
for _, e := range all {
if e.At < from {
ex.Before.Assets[e.ItemID] += e.Delta
if ex.Before.Assets[e.ItemID] == 0 {
delete(ex.Before.Assets, e.ItemID)
}
ex.Before.Seq = e.Seq
} else if e.At <= to {
ex.Events = append(ex.Events, e)
}
}
// Copy the map: After must not alias Before, or applying the window would
// silently rewrite the "before" picture too.
ex.After = State{Assets: map[uint32]int64{}, Seq: ex.Before.Seq}
for k, v := range ex.Before.Assets {
ex.After.Assets[k] = v
}
for _, e := range ex.Events {
ex.After.Assets[e.ItemID] += e.Delta
if ex.After.Assets[e.ItemID] == 0 {
delete(ex.After.Assets, e.ItemID)
}
ex.After.Seq = e.Seq
}
return ex
}
Replay 里有个容易被忽略的细节:数量减到 0 就把键删掉,而不是留一个值为 0 的条目。这样「背包里有哪些物品」直接看 map 的键就行,不用过滤零值;也让两次不同的操作序列(先加后减、或压根没动过)重放出完全相同的状态,避免「逻辑等价但结构不等」的假差异。
Explain 里那句「Copy the map」注释是踩过坑才加的:Go 的 map 是引用类型,ex.After = ex.Before 会让两个字段指向同一张 map,接着往 After 里叠加窗口事件,Before 也被改了——「变之前」和「变之后」变成一样,审计就废了。修法是显式复制一份 map 再叠加。这类别名 bug 在事件溯源里特别隐蔽,因为状态通常都是 map/slice。
审计查询:回答「为什么变了」
Explain 就是审计查询的骨架。它要回答的不是「现在有什么」,而是「在 [from, to] 这段时间里,这个玩家的资产为什么从 Before 变成了 After」。实现分三步:先把窗口之前的事件折叠出 Before,再把窗口内的事件收进 Events,最后用 Events 把 Before 推进成 After。
有了这三样,客服系统就能拼出一句人话:「该玩家在 10 月 2 日,先是跨服结算 +1 件(单号 order-A),紧接着又 +1 件(同一单号,疑似重复发放),随后回滚 -1 件,净增 1 件。」——Ref 字段把相关事件串成一条因果链,「同一单号出现两次」立刻暴露了重复发放。
这里要强调一个设计取舍:审计查询只读事件流,绝不读当前状态表。状态表可能已经被后续操作覆盖,也可能被脏写污染;只有 append-only 的事件流是可信的。哪怕状态表显示玩家有 1 件,只要事件流重放出 2 件,就说明状态表错了——事件流是事实来源(source of truth),状态表只是它的一个投影。
审计日志还有一条纪律:事件只增不改。发现写错了,不是去改那条事件,而是追加一条更正事件(比如 rollback),让历史保持完整。上面 seed 里的 rollback 就是这种「追加更正」——它不删掉之前那次 grant,而是再记一笔把它抵掉,于是「为什么变了」有完整的故事可读。
验证:重放对齐快照、窗口解释
下面这段是 L1,go test 可以直接跑。它造了一天的资产事件,验证两件事:重放结果与快照一致、窗口解释能列出当天的每一笔增减。
// file: internal/saga/audit/audit_test.go
// tier: L1
// verify: go test ./internal/saga/audit/ -v
package audit
import "testing"
// One day in milliseconds, used to build a readable timeline.
const day = int64(86_400_000)
func seed(s *Store, pid uint64) {
// Morning: a battle reward, then a purchase. Afternoon: a cross-server
// settlement and a rollback of a duplicated grant.
s.Append(Event{PlayerID: pid, Type: "grant", ItemID: 7001, Delta: 3, At: 1 * day, Ref: "battle-1"})
s.Append(Event{PlayerID: pid, Type: "spend", ItemID: 7001, Delta: -1, At: 1*day + 1_000, Ref: "shop-9"})
s.Append(Event{PlayerID: pid, Type: "cross_settle", ItemID: 7002, Delta: 1, At: 2 * day, Ref: "order-A"})
s.Append(Event{PlayerID: pid, Type: "grant", ItemID: 7002, Delta: 1, At: 2*day + 5_000, Ref: "order-A"})
s.Append(Event{PlayerID: pid, Type: "rollback", ItemID: 7002, Delta: -1, At: 2*day + 6_000, Ref: "order-A"})
}
func TestReplayMatchesSnapshot(t *testing.T) {
s := NewStore()
seed(s, 10)
snap := s.Snapshot(10, 2*day+7_000)
full := Replay(s.Events(10, 0, ^uint64(0)))
if full.Assets[7001] != 2 || full.Assets[7002] != 1 {
t.Fatalf("replay assets=%v want 7001:2 7002:1", full.Assets)
}
if snap.Assets[7001] != full.Assets[7001] || snap.Assets[7002] != full.Assets[7002] {
t.Fatalf("snapshot %v != replay %v", snap.Assets, full.Assets)
}
// Replaying twice is deterministic.
if again := Replay(s.Events(10, 0, ^uint64(0))); again.Assets[7002] != full.Assets[7002] {
t.Fatalf("replay not deterministic: %v vs %v", again.Assets, full.Assets)
}
}
func TestExplainAnswersWhyAssetsChanged(t *testing.T) {
s := NewStore()
seed(s, 10)
// "Why did this player's items change on day 2?"
ex := s.Explain(10, 2*day, 2*day+day)
if ex.Before.Assets[7001] != 2 || ex.Before.Assets[7002] != 0 {
t.Fatalf("before=%v want 7001:2 7002:0", ex.Before.Assets)
}
if len(ex.Events) != 3 {
t.Fatalf("day-2 events=%d want 3", len(ex.Events))
}
if ex.Events[0].Ref != "order-A" || ex.Events[2].Type != "rollback" {
t.Fatalf("events out of order: %+v", ex.Events)
}
// Net effect of day 2: +1 of 7002 (settle +grant then -1 rollback).
if ex.After.Assets[7002] != 1 {
t.Fatalf("after=%v want 7002:1", ex.After.Assets)
}
}
第一个测试验证的是事件溯源的自洽性:Replay 全部事件得到的状态,必须和 Snapshot 记下的状态一致。这条断言看似平凡,却是整个系统的护栏——一旦有人绕过事件流直接改状态表,或某条事件在落盘时丢了,快照和重放就会分叉,测试立刻红。第二个测试验证审计查询:窗口是第 2 天,Before 里 7002 还是 0、7001 是 2(第 1 天的战斗奖励减掉一次购买),窗口内有 3 条事件,After 里 7002 净增 1。三笔事件把「为什么变了」讲得清清楚楚。
把两节串起来看:对账(10.2)负责发现差异,事件溯源(10.3)负责解释差异,Saga(10.1)负责在差异发生前尽量减少它。三者各管一段:Saga 尽量别出错,对账兜住漏网的错,事件溯源让人能查清每一个错是怎么发生的。
本节小结 / 实测结论
- 事件溯源存「发生的事」而非「结果」:事件流 append-only,是唯一的事实来源,状态表只是它的投影。
- 快照是缓存不是真相:任何时刻「重放全部事件」的结果必须等于快照,不等就是丢了事件或被脏写。
Replay必须是纯函数,且「减到 0 删键」,让逻辑等价的操作序列重放出结构完全相同的状态。Explain回答「某段时间资产为什么变」:先折叠出窗口前状态,再收窗口内事件,最后推进出窗口后状态。- Go 的 map 是引用类型,
After = Before会别名——必须显式复制,否则「变之前」和「变之后」被写成一团。 - 写错了不改事件,而是追加一条更正事件(如
rollback),让历史永远可读。
阅读导航:上一节:10.2 对账与数据修复工单 · 下一节:11.1 跨服战场端到端 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。