《游戏服务端架构·规模化、全球化与成本》6.1 队列选型与投递语义

produced -> pending -> delivered -> acked
                         |
                         | nack
                         v
                     retrying --(预算耗尽)--> dead-letter

// file: internal/queuecontract/queuecontract.go
// tier: L1
// verify: go test ./internal/queuecontract/
package queuecontract

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

// Semantics 是主题对消费者做出的投递承诺。
type Semantics string

const (
	AtMostOnce  Semantics = "at-most-once"
	AtLeastOnce Semantics = "at-least-once"
)

// Contract 是单个主题的生产者/消费者协议。
type Contract struct {
	Topic        string
	Semantics    Semantics
	MaxAttempts  int
	DeadLetter   string
	KeyField     string
	VersionField string
}

// Validate 在发布端启动时调用，避免运行到一半才发现契约不完整。
func (c Contract) Validate() error {
	switch {
	case c.Topic == "":
		return errors.New("queuecontract: empty topic")
	case c.Semantics != AtMostOnce && c.Semantics != AtLeastOnce:
		return fmt.Errorf("queuecontract: unknown semantics %q", c.Semantics)
	case c.Semantics == AtLeastOnce && c.MaxAttempts < 1:
		return errors.New("queuecontract: at-least-once needs MaxAttempts >= 1")
	case c.Semantics == AtLeastOnce && c.DeadLetter == "":
		return errors.New("queuecontract: at-least-once needs a dead-letter topic")
	case c.KeyField == "":
		return errors.New("queuecontract: empty key field")
	case c.VersionField == "":
		return errors.New("queuecontract: empty version field")
	}
	return nil
}

// IdempotencyOwner 说明重复投递由谁吸收。
func (c Contract) IdempotencyOwner() string {
	if c.Semantics == AtLeastOnce {
		return "consumer"
	}
	return "none"
}

// Message 是线上信封；Attempt 由 broker 维护，生产者不得自行填写。
type Message struct {
	ID      string
	Key     string
	Version int
	Payload []byte
	Headers map[string]string
	Attempt int
}

// Validate 只检查信封，不解释业务负载。
func (m Message) Validate() error {
	switch {
	case m.ID == "":
		return errors.New("queuecontract: empty message id")
	case m.Key == "":
		return errors.New("queuecontract: empty partition key")
	case m.Version < 1:
		return errors.New("queuecontract: schema version must be >= 1")
	case m.Attempt < 0:
		return errors.New("queuecontract: negative attempt")
	}
	return nil
}

// Broker 是测试用的内存替身；生产替换成真实客户端，Ack/Nack 契约不变。
type Broker struct {
	contract Contract
	mu       sync.Mutex
	pending  []Message
	dead     []Message
	acked    int
}

// NewBroker 拒绝不完整的契约，让配置错误在启动期暴露。
func NewBroker(c Contract) (*Broker, error) {
	if err := c.Validate(); err != nil {
		return nil, err
	}
	return &Broker{contract: c}, nil
}

// Contract 返回 broker 持有的契约副本。
func (b *Broker) Contract() Contract { return b.contract }

// Publish 入队一条消息。
func (b *Broker) Publish(m Message) error {
	if err := m.Validate(); err != nil {
		return err
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.pending = append(b.pending, m)
	return nil
}

// Next 取出最老的消息，调用方必须随后 Ack 或 Nack。
func (b *Broker) Next() (Message, bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.pending) == 0 {
		return Message{}, false
	}
	m := b.pending[0]
	b.pending = b.pending[1:]
	return m, true
}

// Ack 确认消息已处理。
func (b *Broker) Ack(Message) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.acked++
}

// Nack 退回消息；达到尝试上限后进入死信。at-most-once 直接丢弃，不重试。
func (b *Broker) Nack(m Message) (deadLettered bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.contract.Semantics == AtMostOnce {
		return false
	}
	m.Attempt++
	if m.Attempt >= b.contract.MaxAttempts {
		b.dead = append(b.dead, m)
		return true
	}
	b.pending = append(b.pending, m)
	return false
}

// DeadLetters 返回死信副本，调用方不能修改 broker 内部状态。
func (b *Broker) DeadLetters() []Message {
	b.mu.Lock()
	defer b.mu.Unlock()
	out := make([]Message, len(b.dead))
	copy(out, b.dead)
	return out
}

// Acked 返回成功确认的条数。
func (b *Broker) Acked() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.acked
}

// Deduper 在有界窗口内抑制重复消息，键是业务消息 ID 而非负载。
type Deduper struct {
	mu   sync.Mutex
	seen map[string]time.Time
	ttl  time.Duration
	now  func() time.Time
}

// NewDeduper 创建窗口去重器；ttl 必须覆盖最大重投间隔。
func NewDeduper(ttl time.Duration) *Deduper {
	return &Deduper{seen: make(map[string]time.Time), ttl: ttl, now: time.Now}
}

// First 判断 ID 是否首次出现并记录。过期项惰性清理，窗口不会无限增长。
func (d *Deduper) First(id string) bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	now := d.now()
	for k, at := range d.seen {
		if now.Sub(at) >= d.ttl {
			delete(d.seen, k)
		}
	}
	if _, ok := d.seen[id]; ok {
		return false
	}
	d.seen[id] = now
	return true
}

cd .local/checks/gsarch-lab
go test -count=1 ./internal/queuecontract/
ok  gsarchlab/internal/queuecontract