从消费幂等到接口幂等
4.2 消息投递语义与消费幂等
处理的是消费者侧的重复投递：键由生产端的事件 ID 提供，状态是 PROCESSING/DONE，租约用于崩溃接管。订单接口面对的重复来源更多：用户双击、客户端超时重试、网关重试、重试放大器（7.2）都可能在几秒内把同一笔下单请求送到服务端。
两者共享同一组原语，但接口幂等多了两件事：请求指纹和响应快照。同一个键配不同参数不能当成重复请求，必须返回 CONFLICT；已经完成的请求要能原样 replay 当时的响应，而不是重新执行业务。
一条幂等记录要保存什么
Record 保存五类信息：请求指纹（参数摘要）、状态、结果快照、处理租约和记录过期时间。状态只有三个：PROCESSING 表示某个实例正在执行，DONE 表示已有终态和可重放的响应，CONFLICT 表示同键不同指纹。租约和过期时间必须分开：租约决定“处理中崩溃后多久可以接管”，过期时间决定“重复窗口有多长”。
// file: internal/platform/idempotency/store.go
// tier: L1
// verify: go test ./internal/platform/idempotency/ -count=1
package idempotency
import (
"context"
"errors"
"sync"
"time"
)
type State string
const (
Processing State = "PROCESSING"
Done State = "DONE"
Conflict State = "CONFLICT"
)
var (
ErrInFlight = errors.New("idempotent request is in flight")
ErrConflict = errors.New("idempotency key reused with different parameters")
)
type Response struct {
Status int
Body []byte
}
type Record struct {
Fingerprint string
State State
Response Response
LeaseUntil time.Time
ExpiresAt time.Time
}
type Store interface {
Begin(ctx context.Context, key, fingerprint string, now time.Time, lease, retention time.Duration) (Record, bool, error)
Complete(ctx context.Context, key, fingerprint string, resp Response, now time.Time, retention time.Duration) error
}
type MemoryStore struct {
mu sync.Mutex
records map[string]Record
Fault func(op string) error
}
func NewMemoryStore() *MemoryStore { return &MemoryStore{records: map[string]Record{}} }
func (s *MemoryStore) Begin(ctx context.Context, key, fingerprint string, now time.Time, lease, retention time.Duration) (Record, bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.Fault != nil {
if err := s.Fault("begin"); err != nil {
return Record{}, false, err
}
}
rec, ok := s.records[key]
if !ok || !now.Before(rec.ExpiresAt) {
rec = Record{Fingerprint: fingerprint, State: Processing, LeaseUntil: now.Add(lease), ExpiresAt: now.Add(retention)}
s.records[key] = rec
return rec, true, nil
}
if rec.Fingerprint != fingerprint {
rec.State = Conflict
return rec, false, nil
}
if rec.State == Done {
return rec, false, nil
}
if now.Before(rec.LeaseUntil) {
return rec, false, nil
}
rec.LeaseUntil = now.Add(lease)
s.records[key] = rec
return rec, true, nil
}
func (s *MemoryStore) Complete(ctx context.Context, key, fingerprint string, resp Response, now time.Time, retention time.Duration) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.Fault != nil {
if err := s.Fault("complete"); err != nil {
return err
}
}
rec, ok := s.records[key]
if !ok || rec.Fingerprint != fingerprint {
return ErrConflict
}
rec.State = Done
rec.Response = resp
rec.LeaseUntil = time.Time{}
rec.ExpiresAt = now.Add(retention)
s.records[key] = rec
return nil
}
Begin 的五个分支对应五种事实：键不存在或已过期，新建并占位；指纹不同，返回冲突；已完成，返回记录供 replay；租约未到期，说明另一个实例还在处理；租约到期，接管并续租。生产实现把 (key) 放进唯一索引，用一条条件 UPDATE 完成“接管或插入”，而不是先查后写。
Guard：先占位，再执行，最后写终态
Guard.Do 把存储操作包成请求级契约。Begin 失败时默认失败关闭：数据库不可用就返回 503，绝不跳过去重直接执行。占位成功后执行业务，成功则写 DONE 和响应快照；业务返回错误则不写终态，等租约到期后走与崩溃相同的恢复路径。
// file: internal/platform/idempotency/guard.go
// tier: L1
// verify: go test ./internal/platform/idempotency/ -count=1
package idempotency
import (
"context"
"errors"
"time"
)
var ErrUnavailable = errors.New("idempotency store unavailable")
type Outcome string
const (
Executed Outcome = "EXECUTED"
Replayed Outcome = "REPLAYED"
InFlight Outcome = "IN_FLIGHT"
Conflicted Outcome = "CONFLICT"
Unavailable Outcome = "UNAVAILABLE"
)
// Guard rejects a mutating request when the store cannot answer instead of
// executing it without dedup protection.
type Guard struct {
Store Store
Lease time.Duration
Retention time.Duration
Now func() time.Time
}
func (g *Guard) Do(ctx context.Context, key, fingerprint string, fn func(context.Context) (Response, error)) (Response, Outcome, error) {
now := time.Now()
if g.Now != nil {
now = g.Now()
}
rec, claimed, err := g.Store.Begin(ctx, key, fingerprint, now, g.Lease, g.Retention)
if err != nil {
return Response{}, Unavailable, ErrUnavailable
}
if !claimed {
switch rec.State {
case Done:
return rec.Response, Replayed, nil
case Conflict:
return Response{}, Conflicted, ErrConflict
default:
return Response{}, InFlight, ErrInFlight
}
}
resp, ferr := fn(ctx)
if ferr != nil {
// No terminal write: retry is allowed after the lease expires, which is
// the same recovery path as a process crash.
return resp, Executed, ferr
}
if err := g.Store.Complete(ctx, key, fingerprint, resp, now, g.Retention); err != nil {
// The side effect may already be visible; the key stays PROCESSING and
// the downstream idempotency key must absorb the retry.
return resp, Executed, err
}
return resp, Executed, nil
}
Complete 失败是本节最危险的窗口：库存已经扣减、支付已经发生，但终态没写。此时 Guard 把错误返回给调用方，记录停在 PROCESSING；租约到期后重试会再次进入 handler。要保证不二次扣款，handler 调用的库存和支付适配器也必须按同一个幂等键去重，这就是“去重表是第一层，下游幂等键是第二层”。
故障注入测试
测试用一个内存账本模拟库存和支付：apply 按幂等键记录预留和扣款，同一个键只生效一次。故障注入分两类：complete 失败模拟“业务成功但状态写失败”，存储整体失败模拟数据库不可用。时间用可控时钟推进，覆盖租约内、租约后和去重窗口后三个时点。
// file: internal/platform/idempotency/guard_test.go
// tier: L1
// verify: go test -v ./internal/platform/idempotency/ -count=1
package idempotency
import (
"context"
"errors"
"sync"
"sync/atomic"
"testing"
"time"
)
var base = time.Date(2026, 10, 19, 4, 0, 0, 0, time.UTC)
type ledger struct {
mu sync.Mutex
stock int
reservations map[string]int
charges map[string]int
}
func newLedger(stock int) *ledger {
return &ledger{stock: stock, reservations: map[string]int{}, charges: map[string]int{}}
}
func (l *ledger) apply(key string, qty, cents int) error {
l.mu.Lock()
defer l.mu.Unlock()
if _, ok := l.reservations[key]; ok {
return nil
}
if l.stock < qty {
return errors.New("OUT_OF_STOCK")
}
l.stock -= qty
l.reservations[key] = qty
l.charges[key] = cents
return nil
}
func newGuard(store Store, clock *time.Time) *Guard {
return &Guard{Store: store, Lease: time.Second, Retention: time.Hour,
Now: func() time.Time { return *clock }}
}
func handler(l *ledger, key string, calls *atomic.Int32, delay time.Duration) func(context.Context) (Response, error) {
return func(context.Context) (Response, error) {
calls.Add(1)
time.Sleep(delay)
if err := l.apply(key, 1, 1000); err != nil {
return Response{Status: 409, Body: []byte("OUT_OF_STOCK")}, nil
}
return Response{Status: 200, Body: []byte("paid")}, nil
}
}
func TestConcurrentSameKeyAndFingerprintConflict(t *testing.T) {
store := NewMemoryStore()
now := base
g := newGuard(store, &now)
l := newLedger(10)
var calls atomic.Int32
const workers = 8
outcomes := make([]Outcome, workers)
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
_, out, err := g.Do(context.Background(), "order-1", "sku=A&qty=1",
handler(l, "order-1", &calls, 20*time.Millisecond))
if err != nil && !errors.Is(err, ErrInFlight) {
t.Errorf("worker %d: %v", i, err)
}
outcomes[i] = out
}(i)
}
wg.Wait()
executed := 0
for _, out := range outcomes {
if out == Executed {
executed++
}
}
if executed != 1 || calls.Load() != 1 || l.stock != 9 || len(l.reservations) != 1 || len(l.charges) != 1 {
t.Fatalf("executed=%d calls=%d stock=%d reservations=%d charges=%d outcomes=%v",
executed, calls.Load(), l.stock, len(l.reservations), len(l.charges), outcomes)
}
_, out, err := g.Do(context.Background(), "order-1", "sku=A&qty=2", handler(l, "order-1", &calls, 0))
if !errors.Is(err, ErrConflict) || out != Conflicted || calls.Load() != 1 {
t.Fatalf("conflict: out=%s err=%v calls=%d", out, err, calls.Load())
}
t.Logf("workers=%d executed=%d calls=%d stock=%d", workers, executed, calls.Load(), l.stock)
}
func TestLeaseTakeoverFaultInjectionAndWindow(t *testing.T) {
store := NewMemoryStore()
now := base
g := newGuard(store, &now)
l := newLedger(10)
var calls atomic.Int32
failures := 1
store.Fault = func(op string) error {
if op == "complete" && failures > 0 {
failures--
return errors.New("store timeout")
}
return nil
}
h := handler(l, "order-2", &calls, 0)
if _, out, err := g.Do(context.Background(), "order-2", "sku=A&qty=1", h); err == nil || out != Executed {
t.Fatalf("first call: out=%s err=%v", out, err)
}
if _, out, err := g.Do(context.Background(), "order-2", "sku=A&qty=1", h); !errors.Is(err, ErrInFlight) || out != InFlight {
t.Fatalf("inside lease: out=%s err=%v", out, err)
}
now = now.Add(2 * time.Second)
if _, out, err := g.Do(context.Background(), "order-2", "sku=A&qty=1", h); err != nil || out != Executed {
t.Fatalf("after lease: out=%s err=%v", out, err)
}
if calls.Load() != 2 || l.stock != 9 || len(l.reservations) != 1 || len(l.charges) != 1 {
t.Fatalf("calls=%d stock=%d reservations=%d charges=%d",
calls.Load(), l.stock, len(l.reservations), len(l.charges))
}
now = now.Add(2 * time.Hour)
if _, out, err := g.Do(context.Background(), "order-2", "sku=A&qty=1", h); err != nil || out != Executed || calls.Load() != 3 {
t.Fatalf("after window: out=%s err=%v calls=%d", out, err, calls.Load())
}
t.Logf("calls=%d stock=%d reservations=%d charges=%d", calls.Load(), l.stock, len(l.reservations), len(l.charges))
store.Fault = func(string) error { return errors.New("connection refused") }
if _, out, err := g.Do(context.Background(), "order-3", "sku=A&qty=1", h); !errors.Is(err, ErrUnavailable) || out != Unavailable || calls.Load() != 3 {
t.Fatalf("store down: out=%s err=%v calls=%d", out, err, calls.Load())
}
}
真实运行：三次执行，一次扣款
本机 macOS darwin/arm64、Go 1.26.0，命令 go test -v ./... -count=1，原始输出在 .local/reports/be-arch/be-arch-perf-07-3-idempotency-tests.txt；-race 单次运行见 .local/reports/be-arch/be-arch-perf-07-all-race.txt。以下是本机单次运行：
=== RUN TestConcurrentSameKeyAndFingerprintConflict
guard_test.go:92: workers=8 executed=1 calls=1 stock=9
--- PASS: TestConcurrentSameKeyAndFingerprintConflict (0.02s)
=== RUN TestLeaseTakeoverFaultInjectionAndWindow
guard_test.go:128: calls=3 stock=9 reservations=1 charges=1
--- PASS: TestLeaseTakeoverFaultInjectionAndWindow (0.00s)
PASS
ok archlab/internal/platform/idempotency 0.249s
第一组：8 个 goroutine 同时用同一个键下单，只有 1 个拿到 EXECUTED，其余在 IN_FLIGHT；库存从 10 变成 9，预留和扣款各 1 条。第二组按时间顺序发生四件事：第一次调用成功执行业务但 Complete 故障，返回错误且记录停在 PROCESSING；租约内重试被拒绝；租约过期后接管并再次执行；Retention 两小时后再来一次，记录已过期，业务第三次执行。三次 handler 调用只产生 1 次库存扣减和 1 次支付，因为下游账本按同一个幂等键去重。这也标出了去重窗口的真实边界：窗口内由幂等表拦截，窗口外只剩下游幂等键。
存储不可用时失败关闭还是失败开放
默认必须失败关闭。Begin 失败意味着系统无法判断这是不是重复请求；此时执行下单，可能直接造成二次扣款，而数据库故障通常伴随重试和流量放大，正是重复最多的时候。失败关闭的代价是可用性下降：幂等库故障时下单接口整体 503，但这是可恢复的，重复扣款不可恢复。
失败开放只适用于满足全部条件的场景：操作本身天然幂等（例如按 order_id 覆盖写）、重复不会造成资金或库存副作用、且调用方有独立对账。只读查询可以失败开放；创建订单、扣库存、支付、退款必须失败关闭。不要把“提高可用性”当成默许失败开放的理由。
常见错误与边界
第一，只存键不存指纹。同键不同金额的请求会被当成重复并 replay 旧响应，正确做法是返回 CONFLICT 让调用方修正。
第二，先执行业务再写幂等记录。进程在两步之间崩溃，重试会再次执行；顺序必须是先占位、再执行、最后写终态。
第三，把租约和去重窗口设成同一个值。租约只需覆盖一次处理，窗口要覆盖客户端和网关的最长重试时间；窗口短于重试时间，旧请求会穿透。
第四，认为幂等表能覆盖外部副作用。支付、短信等不可回滚操作仍可能“成功但状态未写”，必须让下游也支持幂等键或对账补偿。
第五，存储故障时静默放行。失败开放必须显式配置、限定接口，并记录指标；默认行为应当是 503。
小结
接口幂等需要键、指纹、三态、结果快照、租约和过期窗口六件套。arch-lab 的实现让 8 个并发同键请求只执行一次；在 Complete 故障注入下，租约接管和窗口过期一共触发三次 handler，但库存与支付各只发生一次，证明去重表与下游幂等键是两层防线。存储不可用时默认失败关闭，只读和天然幂等操作才允许例外。下一章从单点故障出发，讨论冗余、故障域与单点消除。
阅读导航：上一节：7.2 重试放大与退避抖动 · 下一节：8.1 冗余、故障域与单点消除 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。