《互联网产品的服务端架构·内容、社交与实时互动》9.3 消息可靠性与去重

// file: internal/delivery/dedupe.go
// tier: L1
// verify: go test ./internal/delivery/

package delivery

import (
	"errors"
	"fmt"
)

var (
	// ErrEmptyMsgID 表示没有可去重的消息标识。空 ID 会把所有空消息
	// 误判成同一条，必须在入口拒绝。
	ErrEmptyMsgID = errors.New("delivery: empty message id")
	// ErrClockBackwards 表示调用方传入的 now 小于窗口中已经记录的时间。
	// 时间倒退会让 TTL 计算失效，不能让这种批次继续写窗口。
	ErrClockBackwards = errors.New("delivery: now moved backwards")
	// ErrNilWindow 表示没有可写入的去重窗口。
	ErrNilWindow = errors.New("delivery: nil window")
)

// Window 是一段时间内的消息去重窗口。Seen 以消息 ID 为键，记录该 ID
// 第一次被接受的时间；TTL 是保留时长，单位与 now 一致。
//
// Window 不是并发安全的。调用方应当在单 goroutine 内使用，或者在外层
// 加锁；否则 Seen 的读写会同时发生。
type Window struct {
	TTL  int64
	Seen map[string]int64
}

// Accept 判断 msgID 在 now 这一刻是否应该被接受。同一条消息在一个 TTL
// 内只接受一次，重复调用返回 false；距离第一次接受达到或超过 TTL 后，
// 同一条消息可以再次被接受，用于客户端合法重发或跨窗口的新一轮投递。
//
// now 必须单调不减。只要窗口里存在比 now 更晚的时间戳，就认为时钟倒退，
// 返回 ErrClockBackwards，并且不写入任何状态。空 msgID 返回 ErrEmptyMsgID。
func Accept(w *Window, msgID string, now int64) (bool, error) {
	if w == nil {
		return false, ErrNilWindow
	}
	if msgID == "" {
		return false, ErrEmptyMsgID
	}
	for seenID, seenAt := range w.Seen {
		if seenAt > now {
			return false, fmt.Errorf("%w: %s seen at %d, now %d", ErrClockBackwards, seenID, seenAt, now)
		}
	}
	if w.Seen == nil {
		w.Seen = make(map[string]int64)
	}
	if prev, ok := w.Seen[msgID]; ok && now-prev < w.TTL {
		return false, nil
	}
	w.Seen[msgID] = now
	return true, nil
}

// Prune 删除已经超过 TTL 的记录，返回清理条数。年龄达到 TTL 的记录
// 视为过期，与 Accept 的判定保持一致。now 之前记录的将来时间戳不会
// 被删除，交给时钟倒退检测处理。
func Prune(w *Window, now int64) int {
	if w == nil {
		return 0
	}
	removed := 0
	for msgID, seenAt := range w.Seen {
		if now-seenAt >= w.TTL {
			delete(w.Seen, msgID)
			removed++
		}
	}
	return removed
}

// file: internal/delivery/dedupe_test.go
// tier: L1
// verify: go test ./internal/delivery/

package delivery

import (
	"errors"
	"testing"
)

func TestAcceptFirstThenDuplicateWithinTTL(t *testing.T) {
	w := &Window{TTL: 60, Seen: map[string]int64{}}
	ok, err := Accept(w, "m1", 1000)
	if err != nil || !ok {
		t.Fatalf("first accept = %v, %v; want true, nil", ok, err)
	}
	ok, err = Accept(w, "m1", 1030)
	if err != nil {
		t.Fatalf("duplicate err = %v", err)
	}
	if ok {
		t.Fatal("duplicate accepted within ttl; want false")
	}
}

func TestAcceptAfterTTLExpires(t *testing.T) {
	w := &Window{TTL: 60, Seen: map[string]int64{}}
	if ok, _ := Accept(w, "m1", 1000); !ok {
		t.Fatal("first accept rejected")
	}
	ok, err := Accept(w, "m1", 1060)
	if err != nil {
		t.Fatalf("accept after ttl err = %v", err)
	}
	if !ok {
		t.Fatal("message still rejected at ttl boundary; want true")
	}
}

func TestAcceptDifferentMessagesIndependent(t *testing.T) {
	w := &Window{TTL: 60, Seen: map[string]int64{}}
	if ok, _ := Accept(w, "m1", 1000); !ok {
		t.Fatal("m1 rejected")
	}
	ok, err := Accept(w, "m2", 1000)
	if err != nil {
		t.Fatalf("m2 err = %v", err)
	}
	if !ok {
		t.Fatal("m2 rejected; want accepted")
	}
}

func TestAcceptClockBackwardsErrors(t *testing.T) {
	w := &Window{TTL: 60, Seen: map[string]int64{}}
	if ok, _ := Accept(w, "m1", 1000); !ok {
		t.Fatal("setup accept failed")
	}
	ok, err := Accept(w, "m2", 900)
	if !errors.Is(err, ErrClockBackwards) {
		t.Fatalf("err = %v; want ErrClockBackwards", err)
	}
	if ok {
		t.Fatal("backwards accept returned true")
	}
	if _, exists := w.Seen["m2"]; exists {
		t.Fatal("backwards accept wrote state")
	}
}

func TestAcceptRejectsEmptyIDAndNilWindow(t *testing.T) {
	if _, err := Accept(&Window{TTL: 60}, "", 1000); !errors.Is(err, ErrEmptyMsgID) {
		t.Fatalf("empty id err = %v; want ErrEmptyMsgID", err)
	}
	if _, err := Accept(nil, "m1", 1000); !errors.Is(err, ErrNilWindow) {
		t.Fatalf("nil window err = %v; want ErrNilWindow", err)
	}
}

func TestPruneRemovesExpiredOnly(t *testing.T) {
	w := &Window{TTL: 60, Seen: map[string]int64{
		"old":    900,
		"edge":   940,
		"fresh":  990,
		"future": 1100,
	}}
	got := Prune(w, 1000)
	if got != 2 {
		t.Fatalf("pruned = %d; want 2", got)
	}
	if _, ok := w.Seen["old"]; ok {
		t.Fatal("old entry not pruned")
	}
	if _, ok := w.Seen["edge"]; ok {
		t.Fatal("boundary entry not pruned")
	}
	if _, ok := w.Seen["fresh"]; !ok {
		t.Fatal("fresh entry pruned")
	}
	if _, ok := w.Seen["future"]; !ok {
		t.Fatal("future entry pruned")
	}
}

func TestPruneNilAndEmpty(t *testing.T) {
	if got := Prune(nil, 1000); got != 0 {
		t.Fatalf("nil prune = %d; want 0", got)
	}
	if got := Prune(&Window{TTL: 60}, 1000); got != 0 {
		t.Fatalf("empty prune = %d; want 0", got)
	}
}

cd .local/checks/webapp-lab && go test ./internal/delivery/ -run 'TestAccept|TestPrune' -count=1 -v

=== RUN   TestAcceptFirstThenDuplicateWithinTTL
--- PASS: TestAcceptFirstThenDuplicateWithinTTL (0.00s)
=== RUN   TestAcceptAfterTTLExpires
--- PASS: TestAcceptAfterTTLExpires (0.00s)
=== RUN   TestAcceptDifferentMessagesIndependent
--- PASS: TestAcceptDifferentMessagesIndependent (0.00s)
=== RUN   TestAcceptClockBackwardsErrors
--- PASS: TestAcceptClockBackwardsErrors (0.00s)
=== RUN   TestAcceptRejectsEmptyIDAndNilWindow
--- PASS: TestAcceptRejectsEmptyIDAndNilWindow (0.00s)
=== RUN   TestPruneRemovesExpiredOnly
--- PASS: TestPruneRemovesExpiredOnly (0.00s)
=== RUN   TestPruneNilAndEmpty
--- PASS: TestPruneNilAndEmpty (0.00s)
PASS
ok  	webapp/internal/delivery	0.359s