一次把配置拉回旧版本的推送
2026 年大促前，arch-lab 订单系统的配置中心同时保留两条下发链路：长连接推送负责秒级生效，实例每 30 秒轮询一次作为兜底。一个实例在 21:14 发生网络分区，恢复后先收到重试队列里的 v85 快照，再收到轮询返回的 v86。它的本地版本原本是 v87。
旧快照没有被拒绝，实例把 v85 当作新配置应用，库存预占 TTL 从 5000 毫秒退回 3000 毫秒；随后 v86 又被接受，水位停在 86。v87 之后的所有推送都被当成“重复版本”丢弃，直到人工重启。大促期间这段窗口产生了重复释放告警。
问题不在推送或轮询本身，而在于两条链路各自做了“版本比较”，规则还不一致：推送只比较消息序号，轮询只比较 HTTP ETag。动态配置的一致性需要一个所有传输共用的裁判，这就是版本水位。分布式配置中心 介绍过配置中心的整体形态，本节只讨论实例侧如何拒绝陈旧数据。
推送与轮询不是二选一
| 维度 | 推送 | 轮询 | 组合策略 |
|---|---|---|---|
| 生效延迟 | 秒级 | 一个轮询周期 | 推送负责快，轮询负责补 |
| 乱序风险 | 高，重试队列会重排 | 低，每次带 since | 统一走版本水位 |
| 故障恢复 | 依赖连接重建和重放 | 下一轮自动追上 | 轮询做反熵 |
| 服务端成本 | 长连接和广播 | 周期性请求 | 按实例规模调周期 |
| 回滚 | 广播旧内容即可 | 下一次轮询拉取 | 回滚也必须是新版本 |
推送的优点是快，缺点是“到达顺序不等于发送顺序”。重试、多副本广播、连接迁移都会让旧消息晚到。轮询的优点是每次都能重新对齐，缺点是延迟和请求量。生产系统通常两者都用，但前提是它们共享同一个应用入口和同一套拒绝规则，否则两条链路会互相覆盖。
版本水位是唯一裁判
水位有三个规则。第一，收到版本低于水位的快照，拒绝并增加 stale 计数。第二，版本等于水位且内容相同，视为重复投递，幂等返回；内容不同则是冲突，必须拒绝并告警，因为同一个版本号不能对应两份内容。第三，版本高于水位才接受，接受前把旧快照放入有界历史，供回滚使用。
回滚是向前发布
回滚的目标是恢复旧内容，不是恢复旧版本号；回滚与兼容窗口
讨论过版本兼容，本节只处理配置快照的版本方向。实现里的 Republish 从有界历史中取出目标版本的 values，生成 watermark + 1 的新快照，再走一次正常的 Apply。这样所有实例看到的仍是一条向前的版本序列，水位不回退，旧版本也不会在之后“复活”。
有界历史决定了能回滚多远。保留 8 个版本适合分钟级变更，保留 100 个版本会占用更多内存，也让“回滚到哪一版”更难判断。历史只保存快照，不保存变更原因，所以生产上还要把版本号与发布单、审计日志关联起来，否则回滚目标仍然靠猜。
可运行实现
下面的代码与 5.1、5.2 共用 config 包。Subscriber 持有当前快照、水位、有界历史和技术计数器；Poller 只是触发 Apply 的一种方式，推送处理器也应直接调用 Apply。
// file: internal/platform/config/sync.go
// tier: L2
// verify: go test ./internal/platform/config/ -count=1
package config
import (
"context"
"errors"
"fmt"
"sync"
)
var (
ErrStaleVersion = errors.New("config: stale version")
ErrVersionConflict = errors.New("config: same version, different content")
ErrUnknownVersion = errors.New("config: unknown version")
)
// Snapshot is one immutable configuration version received from a source.
type Snapshot struct {
Version uint64
Values map[string]string
}
func (s Snapshot) Validate() error {
if s.Version == 0 {
return errors.New("config: snapshot version must be positive")
}
return nil
}
func (s Snapshot) Clone() Snapshot {
return Snapshot{Version: s.Version, Values: cloneValues(s.Values)}
}
// Equal treats the same version with different values as a conflict.
func (s Snapshot) Equal(other Snapshot) bool {
if s.Version != other.Version || len(s.Values) != len(other.Values) {
return false
}
for k, v := range s.Values {
if ov, ok := other.Values[k]; !ok || ov != v {
return false
}
}
return true
}
type ApplyResult struct {
Applied bool
Version uint64
Reason string
}
// Subscriber owns one node's config and watermark; push and poll share Apply.
type Subscriber struct {
mu sync.RWMutex
current Snapshot
watermark uint64
history []Snapshot
maxHistory int
applied uint64
stale uint64
conflicts uint64
}
// NewSubscriber starts at a known-good snapshot.
func NewSubscriber(initial Snapshot, maxHistory int) (*Subscriber, error) {
if err := initial.Validate(); err != nil {
return nil, err
}
if maxHistory < 1 {
maxHistory = 8
}
return &Subscriber{current: initial.Clone(), watermark: initial.Version, maxHistory: maxHistory}, nil
}
// Apply accepts a newer version, ignores identical duplicates, and rejects
// anything that would move the watermark backwards or rewrite history.
func (s *Subscriber) Apply(next Snapshot) (ApplyResult, error) {
if err := next.Validate(); err != nil {
return ApplyResult{}, err
}
s.mu.Lock()
defer s.mu.Unlock()
return s.applyLocked(next)
}
func (s *Subscriber) applyLocked(next Snapshot) (ApplyResult, error) {
switch {
case next.Version < s.watermark:
s.stale++
return ApplyResult{Version: s.watermark, Reason: "stale"},
fmt.Errorf("%w: got %d, watermark %d", ErrStaleVersion, next.Version, s.watermark)
case next.Version == s.watermark:
if s.current.Equal(next) {
return ApplyResult{Version: s.watermark, Reason: "duplicate"}, nil
}
s.conflicts++
return ApplyResult{Version: s.watermark, Reason: "conflict"},
fmt.Errorf("%w: version %d", ErrVersionConflict, next.Version)
}
s.history = append(s.history, s.current)
if len(s.history) > s.maxHistory {
s.history = s.history[len(s.history)-s.maxHistory:]
}
s.current = next.Clone()
s.watermark = next.Version
s.applied++
return ApplyResult{Applied: true, Version: next.Version, Reason: "applied"}, nil
}
func (s *Subscriber) Current() Snapshot {
s.mu.RLock()
defer s.mu.RUnlock()
return s.current.Clone()
}
func (s *Subscriber) Watermark() uint64 {
s.mu.RLock()
defer s.mu.RUnlock()
return s.watermark
}
// Republish rolls back by publishing old values under a NEW version. Rewinding
// the watermark would make every later update look stale.
func (s *Subscriber) Republish(version uint64) (Snapshot, error) {
s.mu.Lock()
defer s.mu.Unlock()
var target *Snapshot
for i := range s.history {
if s.history[i].Version == version {
target = &s.history[i]
break
}
}
if target == nil {
return Snapshot{}, fmt.Errorf("%w: %d", ErrUnknownVersion, version)
}
next := Snapshot{Version: s.watermark + 1, Values: cloneValues(target.Values)}
if _, err := s.applyLocked(next); err != nil {
return Snapshot{}, err
}
return next.Clone(), nil
}
// Source is the server side; modified=false means since is current.
type Source interface {
Fetch(ctx context.Context, since uint64) (Snapshot, bool, error)
}
type Poller struct {
sub *Subscriber
source Source
}
func NewPoller(sub *Subscriber, source Source) (*Poller, error) {
if sub == nil || source == nil {
return nil, errors.New("config: poller needs a subscriber and a source")
}
return &Poller{sub: sub, source: source}, nil
}
// PollOnce applies a fetched payload through the same checks as push.
func (p *Poller) PollOnce(ctx context.Context) (ApplyResult, error) {
since := p.sub.Watermark()
snap, modified, err := p.source.Fetch(ctx, since)
if err != nil {
return ApplyResult{}, fmt.Errorf("config: poll: %w", err)
}
if !modified {
return ApplyResult{Version: since, Reason: "not modified"}, nil
}
return p.sub.Apply(snap)
}
实现里有两个关键取舍。第一，历史保存的是完整快照而不是差异，回滚时不必重放变更链，代价是内存随版本数和配置大小增长。第二，Apply 返回结构化结果而不是只返回 error，调用方可以把 stale、conflict、duplicate 变成指标，否则拒绝会静默发生。
本机实测
测试覆盖水位拒绝、重复与冲突的区分、回滚生成新版本、轮询的 since 传递和旧快照拒绝。命令与环境如下，原始输出保存在 .local/reports/be-arch-gov/ch05-be-arch-gov-05-3-dynamic-config-push-poll-rollback.txt。
$ go version
go version go1.26.0 darwin/arm64
$ gofmt -l .
$ go vet ./...
$ go test ./internal/platform/config/ -count=1 -v -run 'TestApply|TestRepublish|TestPoller'
=== RUN TestApplyMonotonic
sync_test.go:33: watermark=2 stale=1 reason=stale
--- PASS: TestApplyMonotonic (0.00s)
=== RUN TestApplyDuplicateAndConflict
sync_test.go:49: duplicate=duplicate conflict=conflict applied=0 conflicts=1
--- PASS: TestApplyDuplicateAndConflict (0.00s)
=== RUN TestRepublishRollback
sync_test.go:71: rolled_version=4 final_watermark=4
--- PASS: TestRepublishRollback (0.00s)
=== RUN TestPoller
sync_test.go:110: calls=3 applied=1 stale=1 watermark=2
--- PASS: TestPoller (0.00s)
PASS
ok archlab/internal/platform/config 0.236s
测试块如下：
// file: internal/platform/config/sync_test.go
// tier: L2
// verify: go test ./internal/platform/config/ -count=1
package config
import (
"context"
"errors"
"testing"
)
func mustSubscriber(t *testing.T, snap Snapshot) *Subscriber {
t.Helper()
sub, err := NewSubscriber(snap, 4)
if err != nil {
t.Fatalf("NewSubscriber: %v", err)
}
return sub
}
func TestApplyMonotonic(t *testing.T) {
sub := mustSubscriber(t, Snapshot{Version: 2, Values: map[string]string{"retry": "3"}})
res, err := sub.Apply(Snapshot{Version: 1, Values: map[string]string{"retry": "2"}})
if !errors.Is(err, ErrStaleVersion) || res.Applied || res.Version != 2 {
t.Fatalf("stale result = %+v, %v", res, err)
}
if got := sub.Current(); got.Version != 2 || got.Values["retry"] != "3" {
t.Fatalf("current changed: %+v", got)
}
if sub.stale != 1 || sub.applied != 0 {
t.Fatalf("stale=%d applied=%d", sub.stale, sub.applied)
}
t.Logf("watermark=%d stale=%d reason=%s", sub.Watermark(), sub.stale, res.Reason)
}
func TestApplyDuplicateAndConflict(t *testing.T) {
sub := mustSubscriber(t, Snapshot{Version: 5, Values: map[string]string{"mode": "stable"}})
dup, err := sub.Apply(Snapshot{Version: 5, Values: map[string]string{"mode": "stable"}})
if err != nil || dup.Applied || dup.Reason != "duplicate" {
t.Fatalf("duplicate = %+v, %v", dup, err)
}
conflict, err := sub.Apply(Snapshot{Version: 5, Values: map[string]string{"mode": "canary"}})
if !errors.Is(err, ErrVersionConflict) || conflict.Applied {
t.Fatalf("conflict = %+v, %v", conflict, err)
}
if got := sub.Current().Values["mode"]; got != "stable" {
t.Fatalf("conflict overwrote current: %s", got)
}
t.Logf("duplicate=%s conflict=%s applied=%d conflicts=%d",
dup.Reason, conflict.Reason, sub.applied, sub.conflicts)
}
func TestRepublishRollback(t *testing.T) {
sub := mustSubscriber(t, Snapshot{Version: 1, Values: map[string]string{"retry": "2"}})
for _, v := range []uint64{2, 3} {
value := string(rune('0' + v))
if _, err := sub.Apply(Snapshot{Version: v, Values: map[string]string{"retry": value}}); err != nil {
t.Fatalf("apply v%d: %v", v, err)
}
}
rolled, err := sub.Republish(1)
if err != nil || rolled.Version != 4 || rolled.Values["retry"] != "2" {
t.Fatalf("rolled = %+v, %v", rolled, err)
}
if _, err := sub.Apply(Snapshot{Version: 3, Values: map[string]string{"retry": "3"}}); !errors.Is(err, ErrStaleVersion) {
t.Fatalf("post-rollback stale err = %v", err)
}
if _, err := sub.Republish(99); !errors.Is(err, ErrUnknownVersion) {
t.Fatalf("unknown target err = %v", err)
}
t.Logf("rolled_version=%d final_watermark=%d", rolled.Version, sub.Watermark())
}
type fakeSource struct {
snap Snapshot
modified bool
calls int
lastSince uint64
}
func (f *fakeSource) Fetch(_ context.Context, since uint64) (Snapshot, bool, error) {
f.calls++
f.lastSince = since
return f.snap, f.modified, nil
}
func TestPoller(t *testing.T) {
sub := mustSubscriber(t, Snapshot{Version: 1, Values: map[string]string{"retry": "2"}})
src := &fakeSource{snap: Snapshot{Version: 2, Values: map[string]string{"retry": "3"}}, modified: true}
poller, err := NewPoller(sub, src)
if err != nil {
t.Fatalf("NewPoller: %v", err)
}
res, err := poller.PollOnce(context.Background())
if err != nil || !res.Applied || res.Version != 2 || src.lastSince != 1 {
t.Fatalf("first poll = %+v, %v, since=%d", res, err, src.lastSince)
}
src.modified = false
res, err = poller.PollOnce(context.Background())
if err != nil || res.Applied || res.Reason != "not modified" || src.lastSince != 2 {
t.Fatalf("second poll = %+v, %v, since=%d", res, err, src.lastSince)
}
src.modified, src.snap = true, Snapshot{Version: 1, Values: map[string]string{"retry": "1"}}
if _, err := poller.PollOnce(context.Background()); !errors.Is(err, ErrStaleVersion) {
t.Fatalf("stale poll err = %v", err)
}
if sub.Watermark() != 2 {
t.Fatalf("watermark = %d", sub.Watermark())
}
t.Logf("calls=%d applied=%d stale=%d watermark=%d",
src.calls, sub.applied, sub.stale, sub.Watermark())
}
读数是四组。水位组，本地 v2 收到 v1 后返回 stale，当前快照和水位都保持 v2。重复与冲突组，同版本同内容返回 duplicate 且不增加 applied，同版本不同内容返回 conflict 且不覆盖当前值。回滚组，v1→v3 后回滚到 v1，生成 v4，水位升到 4，随后到达的 v3 被拒绝。轮询组，第一次以 since=1 拉到 v2，第二次 since=2 返回 not modified，第三次收到 v1 被拒绝，水位仍是 2。
边界要说清楚：这是单进程、内存快照、固定历史的测试，不覆盖真实网络分区、配置中心多副本和时钟漂移。生产上水位只能保证单实例单调，不能保证所有实例同时到达同一版本；跨实例的一致性需要靠发布系统和观测窗口判断。0.24 秒是本机测试耗时，不代表真实推送延迟。
常见错误与边界
- 推送和轮询各写一套版本判断。规则一旦分叉，两条链路就会互相覆盖；所有写入必须经过同一个
Apply。 - 把“同版本不同内容”当成重复投递。接受它会破坏幂等，拒绝它才能让版本号保持唯一。
- 回滚时直接改回水位。水位回拨会让后续合法更新永久被拒；回滚应发布旧内容的新版本。
- 历史无限增长。完整快照占用内存，必须有界，并把版本号与发布单关联。
小结
动态配置的一致性不取决于用推送还是轮询，而取决于所有传输是否共享同一个版本水位。低于水位的快照丢弃，等于水位的同内容幂等、异内容拒绝，高于水位才接受；回滚则把旧内容作为新版本重新发布，保持水位单调。
本节用 Subscriber 把这三条规则固定下来，用 Poller 证明轮询只是 Apply 的一种触发方式。配合 5.1 的分层版本和 5.2 的开关生命周期，配置、开关和运行时同步才构成一条可审计的治理链；下一章转向可观测性，讨论这些变更如何被日志、指标和追踪统一记录。
阅读导航：上一节：5.2 功能开关：类型、生命周期与开关债 · 下一节：6.1 日志、指标与追踪的统一语义约定 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。