《MMO 服务器开发》7.3 时钟回拨与容错

// file: internal/idgen/clockback.go
// tier: L1
// verify: go test ./internal/idgen/
package idgen

import (
	"fmt"
	"sync"
	"time"
)

// Strategy 是时钟回拨时的应对策略。
type Strategy int

const (
	// StrategyWait 自旋等待本地时钟追平上次生成时刻，超过容忍上限则报错。
	StrategyWait Strategy = iota
	// StrategyBorrowWorker 从备用 worker 池借一个 worker id，换身份继续生成。
	StrategyBorrowWorker
	// StrategyReject 直接拒绝，返回包装了 ErrClockBackwards 的错误。
	StrategyReject
)

// TolerantSnowflake 在雪花算法之上加了回拨容错。位布局与 Snowflake 完全一致，
// 因此两种实现生成的 ID 可以混在同一个空间里。
type TolerantSnowflake struct {
	mu           sync.Mutex
	clock        Clock
	workerID     int64
	spareWorkers []int64
	strategy     Strategy
	maxBackward  int64 // 允许等待的最大回拨毫秒数
	lastMs       int64
	sequence     int64
	borrowed     int64 // 已借用的 worker 次数，供观测
}

// TolerantOption 是可选配置。
type TolerantOption func(*TolerantSnowflake)

// WithStrategy 设置回拨策略。
func WithStrategy(s Strategy) TolerantOption {
	return func(g *TolerantSnowflake) { g.strategy = s }
}

// WithMaxBackward 设置等待策略能容忍的最大回拨毫秒数。
func WithMaxBackward(ms int64) TolerantOption {
	return func(g *TolerantSnowflake) { g.maxBackward = ms }
}

// WithSpareWorkers 设置可借用的备用 worker id 池（仅 StrategyBorrowWorker 用）。
func WithSpareWorkers(ids ...int64) TolerantOption {
	return func(g *TolerantSnowflake) { g.spareWorkers = append([]int64(nil), ids...) }
}

// NewTolerantSnowflake 构造带回拨容错的生成器。
func NewTolerantSnowflake(workerID int64, clock Clock, opts ...TolerantOption) (*TolerantSnowflake, error) {
	if workerID < 0 || workerID > MaxWorkerID {
		return nil, ErrWorkerRange
	}
	if clock == nil {
		clock = RealClock{}
	}
	g := &TolerantSnowflake{clock: clock, workerID: workerID, strategy: StrategyWait, maxBackward: 5}
	for _, o := range opts {
		o(g)
	}
	return g, nil
}

// WorkerID 返回当前生效的 worker id（借位后可能已改变）。
func (g *TolerantSnowflake) WorkerID() int64 {
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.workerID
}

// Borrowed 返回累计借位次数。
func (g *TolerantSnowflake) Borrowed() int64 {
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.borrowed
}

// Next 生成下一个 ID，并在时钟回拨时按配置的策略恢复。
func (g *TolerantSnowflake) Next() (uint64, error) {
	g.mu.Lock()
	defer g.mu.Unlock()

	now := g.clock.NowMillis()
	if now < g.lastMs {
		fixed, err := g.recover(now)
		if err != nil {
			return 0, err
		}
		now = fixed
	}
	if now == g.lastMs {
		g.sequence = (g.sequence + 1) & MaxSequence
		if g.sequence == 0 {
			for now <= g.lastMs {
				time.Sleep(100 * time.Microsecond)
				now = g.clock.NowMillis()
			}
		}
	} else {
		g.sequence = 0
	}
	g.lastMs = now
	return compose(now, g.workerID, g.sequence), nil
}

func (g *TolerantSnowflake) recover(now int64) (int64, error) {
	switch g.strategy {
	case StrategyReject:
		return 0, fmt.Errorf("%w: last=%d now=%d", ErrClockBackwards, g.lastMs, now)
	case StrategyWait:
		back := g.lastMs - now
		if back > g.maxBackward {
			return 0, fmt.Errorf("%w: 回拨 %dms 超过容忍上限 %dms", ErrClockBackwards, back, g.maxBackward)
		}
		for now < g.lastMs {
			time.Sleep(100 * time.Microsecond)
			now = g.clock.NowMillis()
		}
		return now, nil
	case StrategyBorrowWorker:
		if len(g.spareWorkers) == 0 {
			return 0, fmt.Errorf("%w: 备用 worker 池已空，无法借位", ErrClockBackwards)
		}
		g.workerID, g.spareWorkers = g.spareWorkers[0], g.spareWorkers[1:]
		g.borrowed++
		g.sequence = 0
		return now, nil
	}
	return 0, ErrClockBackwards
}

// file: internal/idgen/clockback_test.go
// tier: L1
// verify: go test ./internal/idgen/ -run TestBackward -v
package idgen

import (
	"errors"
	"sync"
	"testing"
)

// fakeClock 是可注入的假时钟：可以任意前后拨动，用来复现回拨场景。
type fakeClock struct {
	mu  sync.Mutex
	now int64
}

func (c *fakeClock) NowMillis() int64 {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.now
}

func (c *fakeClock) Set(v int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.now = v
}

func (c *fakeClock) Advance(d int64) { c.Set(c.NowMillis() + d) }

// 成因一：NTP 校时——本地时钟被往回拨。拒绝策略直接报错。
func TestBackwardReject(t *testing.T) {
	clk := &fakeClock{now: Epoch + 10_000}
	g, err := NewTolerantSnowflake(9, clk, WithStrategy(StrategyReject))
	if err != nil {
		t.Fatalf("new: %v", err)
	}
	if _, err := g.Next(); err != nil {
		t.Fatalf("first: %v", err)
	}
	clk.Advance(-5000) // NTP 把时钟往回拨 5 秒
	if _, err := g.Next(); !errors.Is(err, ErrClockBackwards) {
		t.Fatalf("回拨后应拒绝生成，got %v", err)
	}
}

// 成因三：闰秒/大幅校时——回退很大且不能等。借位换 worker 继续生成。
func TestBackwardBorrowWorker(t *testing.T) {
	clk := &fakeClock{now: Epoch + 10_000}
	g, err := NewTolerantSnowflake(9, clk,
		WithStrategy(StrategyBorrowWorker), WithSpareWorkers(11, 12))
	if err != nil {
		t.Fatalf("new: %v", err)
	}
	first, _ := g.Next()
	if WorkerIDOf(first) != 9 {
		t.Fatalf("初始 worker 应为 9，got %d", WorkerIDOf(first))
	}
	clk.Advance(-60_000) // 回退一分钟，借位策略不等待
	second, err := g.Next()
	if err != nil {
		t.Fatalf("借位策略应成功：%v", err)
	}
	if got := WorkerIDOf(second); got != 11 {
		t.Fatalf("借位后 worker 应为 11，got %d", got)
	}
	if second == first {
		t.Fatal("借位后生成的 ID 不能与旧 ID 相同")
	}
	if g.Borrowed() != 1 {
		t.Fatalf("借位次数应为 1，got %d", g.Borrowed())
	}
	// 第二次回拨，再借一个
	clk.Advance(-60_000)
	third, err := g.Next()
	if err != nil {
		t.Fatalf("第二次借位应成功：%v", err)
	}
	if WorkerIDOf(third) != 12 {
		t.Fatalf("第二次借位后 worker 应为 12，got %d", WorkerIDOf(third))
	}
}

--- PASS: TestBackwardReject (0.00s)
--- PASS: TestBackwardBorrowWorker (0.00s)
--- PASS: TestBackwardWaitCatchesUp (0.02s)
--- PASS: TestBackwardWaitOverLimit (0.00s)
--- PASS: TestBackwardBorrowExhausted (0.00s)