分支抽象不是再加一个接口
站内 模块化单体架构
和 4.2 已经说明端口与适配器如何固定依赖方向，这里不重复。迁移期多出来的问题是：新旧两套实现要同时存在一段时间，调用方不应该知道自己在跟哪一套说话。分支抽象就是这层“同时存在”的契约，它必须比两套实现都窄。如果接口把旧系统的所有方法都抄过来，协调器迟早会开始判断“这个方法旧的有、新的没有”，分支边界就漏了。arch-lab 只保留 Reserve 与 Get：写一次、读一次，刚好够验证迁移；Release 走完全相同的顺序，代码从略。
双写的顺序不是对称的
正确的顺序是：主写成功 → 落重放意图 → 写影子 → 确认意图。反过来先写影子，一旦主写失败回滚，影子侧就留下一笔用户以为不存在的预留，也就是幽灵库存；先写主库则最坏只是影子暂时落后，而落后可以重放。“落意图”这一步不能省。进程如果在主写成功、影子未写之间崩溃，没有意图记录就永远不知道要补哪一笔。示例用进程内切片保存意图；生产实现必须把它换成与主写同事务的 outbox，否则崩溃窗口依然存在；这一点是双写与普通重试的分界线。影子失败不能让用户请求失败，因为主库已经提交，报错只会诱导用户重试并放大问题。但“不影响用户”不等于“不用管”：每一次影子失败都要计数并刷新静默期，退出判据会用它阻止迁移收尾。否则双写会稳定地跑下去，直到某天对账才发现两侧早就分叉。
双读比对：先归一化，再分类
双读永远返回主库结果，影子结果只用于比较。比较之前必须先归一化：旧系统可能存小写 SKU、带空格的状态，新系统统一大写；这些不是数据差异，而是口径差异。归一化函数要显式写出来并被测试覆盖，不能靠“看起来一样”。差异分两类。missing_in_shadow 表示影子侧根本没有这条记录，通常意味着历史数据还没回填；field_mismatch 表示记录存在但状态或数量不同，通常意味着某次写没有同步。影子查询报错不算数据差异，它是观测故障，单独计入静默期。把三者混成一个“差异率”，排查时无法区分该补数据还是该修代码。差异列表会随时间累积，早期差异会稀释近期差异率。生产实现应按时间窗口统计并保留可追溯的样本；示例保留全量列表，是为了让测试能直接断言分类，而不是鼓励把内存当审计日志。
代码
三个文件：领域侧契约、双写协调器、协调器测试。
// file: internal/inventory/inventory.go
// tier: L1
// verify: go test ./internal/migration/
package inventory
import (
"context"
"errors"
)
var ErrNotFound, ErrUnavailable = errors.New("not found"), errors.New("unavailable")
type Reservation struct {
ID, SKU, Status string
Quantity int
}
type Request struct {
ReservationID, SKU string
Quantity int
}
// Branch 是新旧两侧都必须满足的最小契约；Get 用于双读比对。
type Branch interface {
Reserve(context.Context, Request) (Reservation, error)
Get(context.Context, string) (Reservation, error)
}
接口刻意不含 Release。释放的双写与预留完全同构，但把它也写进来会让示例多出一整条测试路径，而本节要证明的是顺序、比对和退出三件事。
// file: internal/migration/dualwrite.go
// tier: L1
// verify: go test ./internal/migration/
package migration
import (
"context"
"errors"
"fmt"
"slices"
"strings"
"sync"
"sync/atomic"
"time"
"archlab/internal/inventory"
)
// PendingWrite 是写入意图。先落意图、再写影子、成功后确认，
// 进程在任意一步崩溃，重放都不会漏掉影子写。
type PendingWrite struct {
Key string
Request inventory.Request
}
type Divergence struct {
ReservationID string
Class string
}
type Stats struct {
Reads, Repairs atomic.Int64
}
type DualWriter struct {
primary, shadow inventory.Branch
readShadow bool
now func() time.Time
stats Stats
mu sync.Mutex
pending []PendingWrite
divergences []Divergence
lastFailure time.Time
}
func NewDualWriter(primary, shadow inventory.Branch, readShadow bool) *DualWriter {
return &DualWriter{primary: primary, shadow: shadow, readShadow: readShadow, now: time.Now}
}
func (d *DualWriter) enqueue(w PendingWrite) {
d.mu.Lock()
defer d.mu.Unlock()
if !slices.ContainsFunc(d.pending, func(x PendingWrite) bool { return x.Key == w.Key }) {
d.pending = append(d.pending, w)
}
}
func (d *DualWriter) ack(key string) {
d.mu.Lock()
defer d.mu.Unlock()
d.pending = slices.DeleteFunc(d.pending, func(w PendingWrite) bool { return w.Key == key })
}
func normalize(r inventory.Reservation) inventory.Reservation {
r.SKU = strings.ToUpper(strings.TrimSpace(r.SKU))
r.Status = strings.ToUpper(strings.TrimSpace(r.Status))
return r
}
// Reserve 的顺序固定：主写成功 → 落重放意图 → 写影子 → 确认。
// 反过来先写影子，一旦主写失败，就会留下用户以为不存在的幽灵库存。
func (d *DualWriter) Reserve(ctx context.Context, req inventory.Request) (inventory.Reservation, error) {
res, err := d.primary.Reserve(ctx, req)
if err != nil {
return inventory.Reservation{}, err
}
w := PendingWrite{Key: req.ReservationID, Request: req}
d.enqueue(w)
if _, err := d.shadow.Reserve(ctx, req); err != nil {
d.fail()
return res, nil
}
d.ack(w.Key)
return res, nil
}
// Get 永远返回主库结果。影子读取只用于比对，失败或差异都不能影响调用方。
func (d *DualWriter) Get(ctx context.Context, reservationID string) (inventory.Reservation, error) {
got, err := d.primary.Get(ctx, reservationID)
if err != nil || !d.readShadow {
return got, err
}
d.stats.Reads.Add(1)
sh, serr := d.shadow.Get(ctx, reservationID)
switch {
case errors.Is(serr, inventory.ErrNotFound):
d.record(Divergence{reservationID, "missing_in_shadow"})
return got, nil
case serr != nil:
d.fail()
return got, nil
}
a, b := normalize(got), normalize(sh)
if a.Status != b.Status || a.Quantity != b.Quantity {
d.record(Divergence{reservationID, "field_mismatch"})
}
return got, nil
}
func (d *DualWriter) record(v Divergence) {
d.mu.Lock()
d.divergences = append(d.divergences, v)
d.mu.Unlock()
}
func (d *DualWriter) fail() {
d.mu.Lock()
d.lastFailure = d.now()
d.mu.Unlock()
}
// Reconcile 按队列顺序重放。影子实现必须按 Key 幂等，
// 否则“写成功但确认前崩溃”会让同一笔预留执行两次。
func (d *DualWriter) Reconcile(ctx context.Context) int {
d.mu.Lock()
items := append([]PendingWrite(nil), d.pending...)
d.mu.Unlock()
fixed := 0
for _, w := range items {
if _, err := d.shadow.Reserve(ctx, w.Request); err != nil {
continue
}
d.ack(w.Key)
d.stats.Repairs.Add(1)
fixed++
}
return fixed
}
type ExitCriteria struct {
MinRepairs int64
MaxDivergenceBps int64
QuietPeriod time.Duration
}
// CanExit 只在“无待重放、差异率达标、已静默”三条同时成立时
// 返回 true，并一次列出全部阻塞原因。
func (d *DualWriter) CanExit(c ExitCriteria) (bool, []string) {
var out []string
d.mu.Lock()
pending, divs := len(d.pending), int64(len(d.divergences))
quiet := d.now().Sub(d.lastFailure) >= c.QuietPeriod
d.mu.Unlock()
if pending != 0 {
out = append(out, fmt.Sprintf("pending shadow writes: %d", pending))
}
if n := d.stats.Repairs.Load(); n < c.MinRepairs {
out = append(out, fmt.Sprintf("reconciled writes %d < %d", n, c.MinRepairs))
}
if reads := d.stats.Reads.Load(); reads == 0 {
out = append(out, "no shadow reads")
} else if bps := divs * 10000 / reads; bps > c.MaxDivergenceBps {
out = append(out, fmt.Sprintf("divergence %d bps > %d", bps, c.MaxDivergenceBps))
}
if !quiet {
out = append(out, "shadow failure inside quiet period")
}
return len(out) == 0, out
}
Reserve 里四步的顺序就是本节的核心。enqueue 按预留 ID 去重，保证同一笔操作重放多次也只有一条意图；Reconcile 依赖影子实现按 Key 幂等，否则“写成功但确认前崩溃”会让同一笔预留执行两次。CanExit 把待重放、对账样本、差异率和静默期放在同一个返回值里，避免发布脚本只检查其中一项。
// file: internal/migration/dualwrite_test.go
// tier: L1
// verify: go test -race ./internal/migration/ -v
package migration
import (
"context"
"strings"
"testing"
"time"
"archlab/internal/inventory"
)
type fake struct {
name string
err error
calls *[]string
items map[string]inventory.Reservation
}
func newFake(name string, calls *[]string) *fake {
return &fake{name: name, calls: calls, items: map[string]inventory.Reservation{}}
}
func (f *fake) Reserve(_ context.Context, req inventory.Request) (inventory.Reservation, error) {
if f.calls != nil {
*f.calls = append(*f.calls, f.name)
}
if f.err != nil {
return inventory.Reservation{}, f.err
}
r := inventory.Reservation{ID: req.ReservationID, SKU: req.SKU, Status: "RESERVED", Quantity: req.Quantity}
f.items[req.ReservationID] = r
return r, nil
}
func (f *fake) Get(_ context.Context, id string) (inventory.Reservation, error) {
r, ok := f.items[id]
if !ok {
return inventory.Reservation{}, inventory.ErrNotFound
}
return r, nil
}
func TestWriteOrderReconcileAndExit(t *testing.T) {
var calls []string
primary, shadow := newFake("primary", &calls), newFake("shadow", &calls)
shadow.err = inventory.ErrUnavailable
d := NewDualWriter(primary, shadow, true)
now := time.Unix(1000, 0)
d.now = func() time.Time { return now }
ctx := context.Background()
_, err := d.Reserve(ctx, inventory.Request{ReservationID: "r-1", SKU: "sku-1", Quantity: 2})
if err != nil {
t.Fatalf("shadow failure must not fail request: %v", err)
}
if got := strings.Join(calls, ","); got != "primary,shadow" {
t.Fatalf("write order = %s", got)
}
criteria := ExitCriteria{MinRepairs: 1, MaxDivergenceBps: 0, QuietPeriod: time.Minute}
if ok, _ := d.CanExit(criteria); ok {
t.Fatal("pending queue must block exit")
}
shadow.err = nil
if n := d.Reconcile(ctx); n != 1 {
t.Fatalf("first reconcile = %d", n)
}
if n := d.Reconcile(ctx); n != 0 {
t.Fatalf("second reconcile = %d", n)
}
if _, err := d.Get(ctx, "r-1"); err != nil {
t.Fatal(err)
}
now = now.Add(2 * time.Minute)
if ok, why := d.CanExit(criteria); !ok {
t.Fatalf("not ready: %v", why)
}
}
func TestDualReadNormalizesAndClassifies(t *testing.T) {
primary, shadow := newFake("p", nil), newFake("s", nil)
d := NewDualWriter(primary, shadow, true)
ctx := context.Background()
primary.items["r-1"] = inventory.Reservation{ID: "r-1", SKU: "sku-1", Status: "RESERVED", Quantity: 1}
shadow.items["r-1"] = inventory.Reservation{ID: "r-1", SKU: " SKU-1 ", Status: "reserved", Quantity: 1}
if _, err := d.Get(ctx, "r-1"); err != nil || len(d.divergences) != 0 {
t.Fatalf("normalize: err=%v divs=%+v", err, d.divergences)
}
shadow.items["r-1"] = inventory.Reservation{ID: "r-1", SKU: "SKU-1", Status: "RESERVED", Quantity: 2}
_, _ = d.Get(ctx, "r-1")
if len(d.divergences) != 1 || d.divergences[0].Class != "field_mismatch" {
t.Fatalf("mismatch: %+v", d.divergences)
}
delete(shadow.items, "r-1")
_, _ = d.Get(ctx, "r-1")
if d.divergences[1].Class != "missing_in_shadow" {
t.Fatalf("missing: %+v", d.divergences)
}
}
两个测试覆盖四个行为：主写先于影子、影子失败进入待重放、对账重放幂等、双读归一化后正确分类；退出判据在同一用例里从阻塞到放行完整走一遍。
何时允许退出双写
三条同时成立才允许：待重放队列为空且连续一段时间为空；对账重放的样本数达到下限；双读差异率低于阈值，并且距离最后一次影子失败超过静默期。少一条都不行——队列空可能只是流量低，差异率低可能只是没读，静默期没到可能只是故障还没暴露。
| 条件 | 示例阈值 | 不满足时说明什么 |
|---|---|---|
| 待重放队列 | 连续 24 小时为 0 | 影子仍在追赶，或有写持续失败 |
| 对账重放 | ≥ 10,000 笔 | 样本不足，幂等性没有被真正验证 |
| 差异率 | ≤ 5 bps | 口径或数据仍有未解释的分叉 |
| 静默期 | 最后一次失败后 7 天 | 故障可能只是暂时没复现 |
退出顺序也有讲究：先把读切到新侧并保留旧侧只读，再停止影子写，最后删除分支与队列。反过来先停影子写，双读就失去了对照；先删旧侧，则回滚无路。双写不是永久架构，它是一段有明确终点的过渡状态。
真实运行结果
以下输出来自本机 macOS（darwin/arm64，Go 1.26.0）：
$ go test -race ./... -count=1 -v
=== RUN TestWriteOrderReconcileAndExit
--- PASS: TestWriteOrderReconcileAndExit (0.00s)
=== RUN TestDualReadNormalizesAndClassifies
--- PASS: TestDualReadNormalizesAndClassifies (0.00s)
ok archlab/internal/migration 1.546s
-race 在这里不是摆设：协调器同时被请求 goroutine 和后台对账 goroutine 调用时，pending 与 divergences 都由同一把锁保护。测试是单线程的，但竞态检测器能抓到未来某次重构把锁去掉的改动。
ADR-009：主写先行，影子异步补齐
备选 A 是先写影子再写主库，否决理由是主库失败会留下幽灵库存，而幽灵库存比“影子少一笔”难处理得多。备选 B 是两侧并行写、任一成功即返回，否决理由是用户看到的结果不再等于权威库，回滚时无法判断哪一侧是真的。备选 C 是把两侧放进同一个分布式事务，否决理由是旧库根本不支持，且引入的耦合比迁移本身更贵。决策：主写同步、影子尽力而为、意图先落、失败重放。后果是写延迟增加一次影子调用（可异步化），存储与运维成本翻倍，但任意时刻用户读到的都是主库事实，且退出条件可以被代码判定。
常见错误与边界
第一，先写影子再写主库。主库失败时影子已经提交，用户重试会再写一次，幽灵库存越积越多。第二，影子失败直接让请求失败。主库已经成功，报错只会诱导重试；正确做法是记录意图、异步补齐，并让退出判据感知到这次失败。
第三，没有幂等键就重放。重放的前提是影子实现按预留 ID 幂等；否则“确认前崩溃”会把同一笔预留执行两次。第四，把双写当终点。双写是过渡态，必须有退出判据和删除计划；没有终点的双写就是永久双倍写入和永久不一致。
小结
分支抽象把“新旧同时存在”收窄成一个最小契约，双写顺序保证用户看到的结果始终等于主库，双读比对用归一化后的分类暴露真实差异，退出判据把“什么时候算迁移完成”变成可执行代码。arch-lab 的顺序是主写、意图、影子、确认；影子可以慢，但不能丢。下一节把这类边界约束进一步固化成适应度函数，让它们在每次提交时自动运行。
阅读导航：上一节：8.1 绞杀者模式与增量迁移 · 下一节：8.3 架构适应度函数：把边界写成测试 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。