AskaEth
99a644d545
state ∈ 有效 / 被拒（需重扫）/ 未知；🔴 -14 是**被拒不是过期**，且**未知原因绝不写成「过期」** （面板文案与错误码都如实）。三态判定为纯函数 + 三态各有用例。 ⚠️ 未做（如实登记）：每账号逐条三态（accounts[] 数组上的三态）本片未做。
396 lines
15 KiB
Go
396 lines
15 KiB
Go
package events
|
||
|
||
// 本文件钉住**账号在线状态迁移**的判定、归一化与落行（design/26 §2.6；切片 E7）。
|
||
//
|
||
// 断言的是**库里的事实**（多了一行没有、那一行的 tier/category/platform_type/summary
|
||
// 是不是预期的），不按日志文本断言 —— 与 store_test.go 同一口径。
|
||
//
|
||
// 覆盖的三类口径，每一类都对应一条设计约束：
|
||
// - 九种 (前态 → 后态) 组合**逐个落行**，包括涉及「未知」的那四种（§2.6 的
|
||
// `known` 三态；§1.1「绝不静默」）；
|
||
// - **没变就不写**（§4.3：账号状态不是逐帧写库的高频遥测，只有迁移是事实）；
|
||
// - **缺平台是调用方的编程错误**（ErrNoPlatform），而**缺通道码只是归一化**
|
||
// （不报错、不丢类）。
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
// accountStatusStoreStub 是只记「落了什么」的测试替身：它让「判定」与「落库」两面
|
||
// 可以分别断言（真库的端到端断言在下面用 newTestStore）。
|
||
type accountStatusStoreStub struct {
|
||
mu sync.Mutex
|
||
rows []Event
|
||
err error
|
||
calls int
|
||
}
|
||
|
||
func (s *accountStatusStoreStub) Record(_ context.Context, ev Event) error {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
s.calls++
|
||
if s.err != nil {
|
||
return s.err
|
||
}
|
||
s.rows = append(s.rows, ev)
|
||
return nil
|
||
}
|
||
|
||
func (s *accountStatusStoreStub) Count(context.Context, Filter) (int64, error) { return 0, nil }
|
||
func (s *accountStatusStoreStub) List(context.Context, Filter) ([]Event, error) {
|
||
return nil, nil
|
||
}
|
||
func (s *accountStatusStoreStub) Counters(context.Context, string) ([]Counter, error) {
|
||
return nil, nil
|
||
}
|
||
func (s *accountStatusStoreStub) Flush(context.Context) error { return nil }
|
||
func (s *accountStatusStoreStub) Sweep(context.Context) (SweepStats, error) {
|
||
return SweepStats{}, nil
|
||
}
|
||
|
||
// recorded 返回已记录的行的副本（避免调用方在锁外读被并发改写）。
|
||
func (s *accountStatusStoreStub) recorded() []Event {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return append([]Event(nil), s.rows...)
|
||
}
|
||
|
||
// sample 构造一条观测样本（平台固定 qq-obv11，账号 id 由用例决定）。
|
||
func sample(account string, online, known bool) StatusSample {
|
||
return StatusSample{
|
||
Platform: PlatformQQ,
|
||
AccountID: account,
|
||
Online: online,
|
||
Known: known,
|
||
Source: StatusSourceProbe,
|
||
}
|
||
}
|
||
|
||
// TestStatusTracker_九种迁移逐个落行：把 (前态 → 后态) 的**全部**组合跑一遍。
|
||
//
|
||
// 为什么逐个列而不是抽样：§2.6 的价值全在「她自己的账号状态**变了**」这一件事上，
|
||
// 而实现里最容易漏的恰恰是「未知 ↔ 确知」那四条边 —— 一个 `if !known { continue }`
|
||
// 就会把它们整类丢掉（这正是 gateway 的 liveAccountOnline 必须缺席、而留存面
|
||
// 必须留痕的那条分界线）。
|
||
func TestStatusTracker_九种迁移逐个落行(t *testing.T) {
|
||
t.Parallel()
|
||
const acct = "bot-1"
|
||
type step struct {
|
||
online bool
|
||
known bool
|
||
}
|
||
for _, tc := range []struct {
|
||
name string
|
||
seq []step
|
||
prev AccountState
|
||
next AccountState
|
||
summary string
|
||
}{
|
||
{"首次→在线", []step{{true, true}}, "", AccountStateOnline, "账号在线（首次确知）"},
|
||
{"首次→离线", []step{{false, true}}, "", AccountStateOffline, "账号离线（首次确知）"},
|
||
{"首次→未知", []step{{false, false}}, "", AccountStateUnknown, "账号状态未知（尚未探到）"},
|
||
{"在线→离线", []step{{true, true}, {false, true}}, AccountStateOnline, AccountStateOffline, "账号离线（在线→离线）"},
|
||
{"离线→在线", []step{{false, true}, {true, true}}, AccountStateOffline, AccountStateOnline, "账号上线（离线→在线）"},
|
||
{"未知→在线", []step{{false, false}, {true, true}}, AccountStateUnknown, AccountStateOnline, "账号上线（未知→在线）"},
|
||
{"未知→离线", []step{{false, false}, {false, true}}, AccountStateUnknown, AccountStateOffline, "账号离线（未知→离线）"},
|
||
{"在线→未知", []step{{true, true}, {false, false}}, AccountStateOnline, AccountStateUnknown, "账号状态未知（在线→未知）"},
|
||
{"离线→未知", []step{{false, true}, {false, false}}, AccountStateOffline, AccountStateUnknown, "账号状态未知（离线→未知）"},
|
||
} {
|
||
t.Run(tc.name, func(t *testing.T) {
|
||
t.Parallel()
|
||
store, _ := newTestStore(t, nil)
|
||
tr := NewStatusTracker(store, testNow)
|
||
ctx := context.Background()
|
||
|
||
// 前几步只负责把状态推到 tc.prev（它们自己也是迁移，照常落行）。
|
||
for i := 0; i < len(tc.seq)-1; i++ {
|
||
if _, err := tr.Observe(ctx, sample(acct, tc.seq[i].online, tc.seq[i].known)); err != nil {
|
||
t.Fatalf("铺垫第 %d 步: %v", i, err)
|
||
}
|
||
}
|
||
last := tc.seq[len(tc.seq)-1]
|
||
res, err := tr.Observe(ctx, sample(acct, last.online, last.known))
|
||
if err != nil {
|
||
t.Fatalf("Observe: %v", err)
|
||
}
|
||
if !res.Changed || res.State != tc.next || res.Previous != tc.prev {
|
||
t.Fatalf("res = {changed:%v state:%s prev:%s}, want {true %s %s}",
|
||
res.Changed, res.State, res.Previous, tc.next, tc.prev)
|
||
}
|
||
if !res.Recorded {
|
||
t.Fatalf("最后一步未落库（Recorded=false）")
|
||
}
|
||
if res.Event.Summary != tc.summary {
|
||
t.Errorf("summary = %q, want %q", res.Event.Summary, tc.summary)
|
||
}
|
||
// 落行进库的字段逐列核对：T2（§2.6 高价值）+ lifecycle（§3.1 的派生关系）
|
||
// + 探活通道码 + 合成 id system 的角色（无当事人的事件）。
|
||
ev := res.Event
|
||
if ev.Tier != TierT2 || ev.Category != CategoryLifecycle {
|
||
t.Errorf("tier/category = %s/%s, want t2/lifecycle", ev.Tier, ev.Category)
|
||
}
|
||
if ev.PlatformType != StatusSourceProbe {
|
||
t.Errorf("platform_type = %q, want %q", ev.PlatformType, StatusSourceProbe)
|
||
}
|
||
if !ev.SelfRelated {
|
||
t.Error("self_related = false, want true（这是她自己的账号）")
|
||
}
|
||
if ev.ActorID != "" || ev.ActorRole != RoleSystem {
|
||
t.Errorf("actor = %q/%q, want \"\"/system（状态由平台侧上报，没有人的当事人）",
|
||
ev.ActorID, ev.ActorRole)
|
||
}
|
||
if ev.ChannelID != "" || ev.MessageID != "" {
|
||
t.Errorf("channel/message = %q/%q, want 空（账号状态不属任何会话或消息）",
|
||
ev.ChannelID, ev.MessageID)
|
||
}
|
||
|
||
// 读回库：最后一步的那一行必须在，且 summary 逐字一致。
|
||
rows, err := store.List(ctx, Filter{Platform: PlatformQQ})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(rows) != len(tc.seq) {
|
||
t.Fatalf("库里有 %d 行, want %d（每一步迁移一行）", len(rows), len(tc.seq))
|
||
}
|
||
if rows[0].Summary != tc.summary {
|
||
t.Errorf("库里最新一行 summary = %q, want %q", rows[0].Summary, tc.summary)
|
||
}
|
||
if rows[0].Tier != TierT2 || !rows[0].SelfRelated {
|
||
t.Errorf("读回的行 = %+v, want t2 + self_related", rows[0])
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// TestStatusTracker_未变化不落行：同一状态连采多次只留**首行**。
|
||
// 这是 §4.3 与 §2.6 的分界线：逐帧写库是纯浪费，而迁移是事实。
|
||
func TestStatusTracker_未变化不落行(t *testing.T) {
|
||
t.Parallel()
|
||
store, _ := newTestStore(t, nil)
|
||
tr := NewStatusTracker(store, testNow)
|
||
ctx := context.Background()
|
||
for i := 0; i < 5; i++ {
|
||
res, err := tr.Observe(ctx, sample("bot-1", true, true))
|
||
if err != nil {
|
||
t.Fatalf("第 %d 次 Observe: %v", i, err)
|
||
}
|
||
if want := i == 0; res.Changed != want {
|
||
t.Fatalf("第 %d 次 Changed = %v, want %v", i, res.Changed, want)
|
||
}
|
||
}
|
||
rows, err := store.List(ctx, Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(rows) != 1 {
|
||
t.Fatalf("库里有 %d 行, want 1（只有首次观测是迁移）", len(rows))
|
||
}
|
||
}
|
||
|
||
// TestStatusTracker_缺平台是编程错误：样本没有平台 ⇒ 包装的 ErrNoPlatform，
|
||
// 且**什么都不写**（既落不进行，也不在迁移表里留一个空键的假状态）。
|
||
func TestStatusTracker_缺平台是编程错误(t *testing.T) {
|
||
t.Parallel()
|
||
store, _ := newTestStore(t, nil)
|
||
tr := NewStatusTracker(store, testNow)
|
||
res, err := tr.Observe(context.Background(), StatusSample{Online: true, Known: true})
|
||
if !errors.Is(err, ErrNoPlatform) {
|
||
t.Fatalf("err = %v, want wrap of ErrNoPlatform", err)
|
||
}
|
||
if res.Changed || res.Recorded {
|
||
t.Errorf("res = %+v, want 未判定也未落库", res)
|
||
}
|
||
if rows, _ := store.List(context.Background(), Filter{}); len(rows) != 0 {
|
||
t.Fatalf("库里 %d 行, want 0", len(rows))
|
||
}
|
||
if states := tr.States(); len(states) != 0 {
|
||
t.Errorf("States() = %v, want 空（坏样本不该污染迁移表）", states)
|
||
}
|
||
}
|
||
|
||
// TestStatusTracker_通道码缺失只是归一化：空通道码不报错、不丢类，
|
||
// 归一成固定兜底码写进 platform_type（§4.1：排障要能分辨「这条是谁记的」）。
|
||
func TestStatusTracker_通道码缺失只是归一化(t *testing.T) {
|
||
t.Parallel()
|
||
store, _ := newTestStore(t, nil)
|
||
tr := NewStatusTracker(store, testNow)
|
||
res, err := tr.Observe(context.Background(), StatusSample{
|
||
Platform: PlatformQQ, AccountID: "bot-1", Online: true, Known: true,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("Observe: %v", err)
|
||
}
|
||
if !res.Recorded {
|
||
t.Fatal("Recorded = false, want true（缺通道码不该阻止落行）")
|
||
}
|
||
if res.Event.PlatformType != statusSourceFallback {
|
||
t.Errorf("platform_type = %q, want %q", res.Event.PlatformType, statusSourceFallback)
|
||
}
|
||
}
|
||
|
||
// TestStatusTracker_未接线只判定不落库：store 为 nil 时照常判迁移（调用方仍能据此
|
||
// 知道「状态变了」），但 Recorded=false —— 绝不假装写过（§4.3 的 fail-visible）。
|
||
func TestStatusTracker_未接线只判定不落库(t *testing.T) {
|
||
t.Parallel()
|
||
tr := NewStatusTracker(nil, testNow)
|
||
res, err := tr.Observe(context.Background(), sample("bot-1", false, true))
|
||
if err != nil {
|
||
t.Fatalf("Observe: %v", err)
|
||
}
|
||
if !res.Changed || res.State != AccountStateOffline {
|
||
t.Errorf("res = %+v, want changed + offline", res)
|
||
}
|
||
if res.Recorded {
|
||
t.Error("Recorded = true, want false（未接线就没有可查的行）")
|
||
}
|
||
}
|
||
|
||
// TestStatusTracker_落库失败_错误上抛且不声称：§4.3 要求逐条留存的失败**必须**让
|
||
// 调用方看见（E3 据此记 WARN）。此时 Recorded 必须为 false —— 否则面板会收到一条
|
||
// 查询面永远查不到的幽灵帧。
|
||
func TestStatusTracker_落库失败_错误上抛且不声称(t *testing.T) {
|
||
t.Parallel()
|
||
boom := errors.New("db 挂了")
|
||
stub := &accountStatusStoreStub{err: boom}
|
||
tr := NewStatusTracker(stub, testNow)
|
||
res, err := tr.Observe(context.Background(), sample("bot-1", true, true))
|
||
if err == nil || !errors.Is(err, boom) {
|
||
t.Fatalf("err = %v, want wrap of %v", err, boom)
|
||
}
|
||
if res.Recorded {
|
||
t.Error("Recorded = true, want false（落库失败了）")
|
||
}
|
||
if !res.Changed {
|
||
t.Error("Changed = false, want true（判定与落库是两件事）")
|
||
}
|
||
if n := len(stub.recorded()); n != 0 {
|
||
t.Errorf("替身收到 %d 行, want 0", n)
|
||
}
|
||
}
|
||
|
||
// TestStatusTracker_并发观测同一迁移只落一行：8 路并发观测同一个账号的同一次迁移。
|
||
// 判定表在临界区内读写，因此只能有一个赢家 —— 多落几行就等于「一次掉线写了 N 行」。
|
||
// （刻意只起 8 个 goroutine：AGENTS §10 的压测纪律要求有界，不堆进程。）
|
||
func TestStatusTracker_并发观测同一迁移只落一行(t *testing.T) {
|
||
t.Parallel()
|
||
store, _ := newTestStore(t, nil)
|
||
tr := NewStatusTracker(store, testNow)
|
||
const n = 8
|
||
var wg sync.WaitGroup
|
||
changed := make([]bool, n)
|
||
wg.Add(n)
|
||
for i := 0; i < n; i++ {
|
||
go func(i int) {
|
||
defer wg.Done()
|
||
res, err := tr.Observe(context.Background(), sample("bot-1", true, true))
|
||
if err != nil {
|
||
t.Errorf("Observe(%d): %v", i, err)
|
||
return
|
||
}
|
||
changed[i] = res.Changed
|
||
}(i)
|
||
}
|
||
wg.Wait()
|
||
winners := 0
|
||
for _, c := range changed {
|
||
if c {
|
||
winners++
|
||
}
|
||
}
|
||
if winners != 1 {
|
||
t.Errorf("Changed 为真的路数 = %d, want 1", winners)
|
||
}
|
||
rows, err := store.List(context.Background(), Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(rows) != 1 {
|
||
t.Fatalf("库里有 %d 行, want 1（同一迁移只该落一行）", len(rows))
|
||
}
|
||
}
|
||
|
||
// TestStatusTracker_多账号互不影响：两个账号各自迁移，行数与账号对应，
|
||
// 绝不因为「同平台」把它们糊成一行。
|
||
func TestStatusTracker_多账号互不影响(t *testing.T) {
|
||
t.Parallel()
|
||
store, _ := newTestStore(t, nil)
|
||
tr := NewStatusTracker(store, testNow)
|
||
ctx := context.Background()
|
||
mustObserve(t, tr, sample("bot-1", true, true))
|
||
mustObserve(t, tr, sample("bot-2", false, true))
|
||
// 第二轮的三个观测都不改变状态 ⇒ 不落行。
|
||
mustObserve(t, tr, sample("bot-1", true, true))
|
||
mustObserve(t, tr, sample("bot-2", false, true))
|
||
rows, err := store.List(ctx, Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(rows) != 2 {
|
||
t.Fatalf("库里有 %d 行, want 2", len(rows))
|
||
}
|
||
ids := map[string]bool{rows[0].AccountID: true, rows[1].AccountID: true}
|
||
if !ids["bot-1"] || !ids["bot-2"] {
|
||
t.Errorf("account_id = %v, want bot-1 与 bot-2 各一行", ids)
|
||
}
|
||
if states := tr.States(); states[PlatformQQ+"/bot-1"] != AccountStateOnline ||
|
||
states[PlatformQQ+"/bot-2"] != AccountStateOffline {
|
||
t.Errorf("States() = %v, want bot-1 在线 / bot-2 离线", states)
|
||
}
|
||
}
|
||
|
||
// TestStatusSummaries_九种组合逐条有文案：文案表必须覆盖全部 (前态, 后态) 组合且每一条
|
||
// 非空 —— 「漏了一种」在这里变红，而不是悄悄落一行空 summary 进库（空摘要等于没留痕）。
|
||
func TestStatusSummaries_九种组合逐条有文案(t *testing.T) {
|
||
t.Parallel()
|
||
table := statusSummaries()
|
||
if n := len(table); n != 9 {
|
||
t.Errorf("文案表 %d 条, want 9（3 种首次观测 + 3×2 种迁移）", n)
|
||
}
|
||
prevStates := []AccountState{"", AccountStateOnline, AccountStateOffline, AccountStateUnknown}
|
||
nextStates := []AccountState{AccountStateOnline, AccountStateOffline, AccountStateUnknown}
|
||
for _, prev := range prevStates {
|
||
for _, next := range nextStates {
|
||
if prev == next {
|
||
continue // 同态不构成迁移
|
||
}
|
||
if s := table[statusSummaryKey(prev, next)]; s == "" {
|
||
t.Errorf("(%q → %q) 没有固定文案，want 一句话常量", prev, next)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// mustObserve 是「这一步必须成功」的简写（失败直接结束用例）。
|
||
func mustObserve(t *testing.T, tr *StatusTracker, s StatusSample) StatusResult {
|
||
t.Helper()
|
||
res, err := tr.Observe(context.Background(), s)
|
||
if err != nil {
|
||
t.Fatalf("Observe(%s/%s): %v", s.Platform, s.AccountID, err)
|
||
}
|
||
return res
|
||
}
|
||
|
||
// TestStatusTracker_时钟零值时由存储层补当前时刻：CreatedAt 留零值（任何真实写入方
|
||
// 都不该自己造时刻），落库后读回来必须是一个**非零**时刻，而不是 1970（1970 会被
|
||
// 保留期立刻扫掉 —— 那正是「刚发生的证据反被丢掉」）。
|
||
func TestStatusTracker_时钟零值时由存储层补当前时刻(t *testing.T) {
|
||
t.Parallel()
|
||
store, _ := newTestStore(t, nil)
|
||
tr := NewStatusTracker(store, func() time.Time { return time.Time{} })
|
||
res := mustObserve(t, tr, sample("bot-1", true, true))
|
||
if !res.Event.CreatedAt.IsZero() {
|
||
t.Fatalf("CreatedAt = %v, want 零值交给存储层填", res.Event.CreatedAt)
|
||
}
|
||
rows, err := store.List(context.Background(), Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(rows) != 1 || rows[0].CreatedAt.IsZero() {
|
||
t.Fatalf("读回的行 = %+v, want 一个非零 created_at", rows)
|
||
}
|
||
}
|