《互联网产品的服务端架构·内容、社交与实时互动》11.3 搜索一致性与删除

// file: internal/search/sync.go
// tier: L1
// verify: go test ./internal/search/

package search

import "fmt"

// Lag 描述两个水位：业务库已提交到的最大时间，和索引已消化到的最大时间。
// 两者都是“源事件时间”，不是处理时间，因此索引水位不应超过业务库水位。
type Lag struct {
	DBUpdatedAt int64
	IndexedAt   int64
}

// Stale 判断索引是否落后超过预算。预算按秒计，0 表示只容许零延迟。
// 时间倒退（索引水位超过业务库水位）视为异常而不是负延迟，
// 因为那通常意味着时钟错乱或消息被伪造，继续计算只会掩盖问题。
func Stale(l Lag, budgetSeconds int64) (bool, error) {
	if budgetSeconds < 0 {
		return false, fmt.Errorf("search: negative budget %d", budgetSeconds)
	}
	if l.DBUpdatedAt < 0 || l.IndexedAt < 0 {
		return false, fmt.Errorf("search: negative timestamp db=%d indexed=%d", l.DBUpdatedAt, l.IndexedAt)
	}
	if l.IndexedAt > l.DBUpdatedAt {
		return false, fmt.Errorf("search: indexed watermark %d is ahead of database %d", l.IndexedAt, l.DBUpdatedAt)
	}
	return l.DBUpdatedAt-l.IndexedAt > budgetSeconds, nil
}

// Retry 按失败列表的顺序挑出待重试文档，未知 ID 直接忽略。
// 保序是为了让重试日志和第一次失败日志能够按行对上，
// 不为了“顺便去重”而改变调用方看到的顺序。
func Retry(changes []Doc, failedIDs []string) []Doc {
	byID := make(map[string]Doc, len(changes))
	for _, change := range changes {
		byID[change.ID] = change
	}
	out := make([]Doc, 0, len(failedIDs))
	for _, id := range failedIDs {
		if doc, ok := byID[id]; ok {
			out = append(out, doc)
		}
	}
	return out
}

// file: internal/search/sync_test.go
// tier: L1
// verify: go test ./internal/search/

package search

import "testing"

func TestStaleBudget(t *testing.T) {
	cases := []struct {
		name    string
		lag     Lag
		budget  int64
		want    bool
		wantErr bool
	}{
		{"caught-up", Lag{100, 100}, 30, false, false},
		{"within-budget", Lag{100, 70}, 30, false, false},
		{"exactly-budget", Lag{100, 70}, 30, false, false},
		{"over-budget", Lag{100, 69}, 30, true, false},
		{"zero-budget-caught-up", Lag{100, 100}, 0, false, false},
		{"zero-budget-lag", Lag{101, 100}, 0, true, false},
		{"indexed-ahead", Lag{100, 101}, 30, false, true},
		{"negative-timestamp", Lag{-1, 0}, 30, false, true},
		{"negative-budget", Lag{100, 100}, -1, false, true},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			got, err := Stale(c.lag, c.budget)
			if (err != nil) != c.wantErr {
				t.Fatalf("Stale(%+v, %d) error = %v, wantErr %v", c.lag, c.budget, err, c.wantErr)
			}
			if err == nil && got != c.want {
				t.Fatalf("Stale(%+v, %d) = %v, want %v", c.lag, c.budget, got, c.want)
			}
		})
	}
}

func TestRetryPreservesFailedOrder(t *testing.T) {
	changes := []Doc{
		{ID: "a", Title: "A"},
		{ID: "b", Title: "B"},
		{ID: "c", Title: "C", Deleted: true},
	}
	got := Retry(changes, []string{"c", "missing", "a"})
	if len(got) != 2 {
		t.Fatalf("Retry() returned %d docs, want 2", len(got))
	}
	if got[0].ID != "c" || got[1].ID != "a" {
		t.Fatalf("Retry() order = %s,%s, want c,a", got[0].ID, got[1].ID)
	}
	if !got[0].Deleted {
		t.Fatal("tombstone should be retried like any other change")
	}
}

cd .local/checks/webapp-lab && go test ./internal/search/ -run 'TestStale|TestRetry' -count=1 -v

=== RUN   TestStaleBudget
=== RUN   TestStaleBudget/caught-up
=== RUN   TestStaleBudget/within-budget
=== RUN   TestStaleBudget/exactly-budget
=== RUN   TestStaleBudget/over-budget
=== RUN   TestStaleBudget/zero-budget-caught-up
=== RUN   TestStaleBudget/zero-budget-lag
=== RUN   TestStaleBudget/indexed-ahead
=== RUN   TestStaleBudget/negative-timestamp
=== RUN   TestStaleBudget/negative-budget
--- PASS: TestStaleBudget (0.00s)
    --- PASS: TestStaleBudget/caught-up (0.00s)
    --- PASS: TestStaleBudget/within-budget (0.00s)
    --- PASS: TestStaleBudget/exactly-budget (0.00s)
    --- PASS: TestStaleBudget/over-budget (0.00s)
    --- PASS: TestStaleBudget/zero-budget-caught-up (0.00s)
    --- PASS: TestStaleBudget/zero-budget-lag (0.00s)
    --- PASS: TestStaleBudget/indexed-ahead (0.00s)
    --- PASS: TestStaleBudget/negative-timestamp (0.00s)
    --- PASS: TestStaleBudget/negative-budget (0.00s)
=== RUN   TestRetryPreservesFailedOrder
--- PASS: TestRetryPreservesFailedOrder (0.00s)
PASS
ok  	webapp/internal/search	0.274s