AskaEth
1ae9158250
P1-1（channelbridge 的 -race 门仍红，红在用例不确定性而不是竞态）： - coalesceTimers 改成**截止时刻**语义：每次 after(d) 记 at = 注入时钟 + d， 推钟只把时钟推进到「当前最早的截止时刻」再触发它。旧写法按 FIFO 逐个触发并 按请求时长累加，实测 12 条消息内攒到 5 个废弃定时器把时钟推 +10s（窗口 2s、 绝对上限 6s）⇒ 循环头走「到点直接发车、不 takeAll」的分支，队列尾部落到下一批。 - 推钟前强制「车道排空」这条不变量（pump/advanceAndFire/waitIdle）： 时钟只表示时间过去，它一转就可能跳过 takeAll，因此先等 worker 把货收进本批。 - sendSettled：投递后按**序号**等「这次投递之后新挂的定时器」（旧的都已被 worker 放弃、再没人监听），供「投递与推钟交替」的用例（每 500ms 一条的绝对上限用例）。 - fireLatest/advanceAndFire 带 done 谓词重试，避免「登记与触发之间续臂」导致 推了钟却没人醒；waitIdle 一边等一边推窗口（worker 停在窗口里时永远走不到退休）。 - 用例侧只补相位与等待，不动断言：TwelveBurst 读历史前等 assistant 行落库； 准入用例断言「最新一次续臂」（原来按下标取 d[1]，隐含臂数假设）； stale 用例两条都投完再推时钟（车道不会在两条之间退休）。 P2-1（替身锁无法用变异证实）选 A：删掉 stubChatter/stubSessions 的锁与访问器， 退回原字段访问，并写清理由 —— 21 个用例全走同步便捷路径 （b.process → processFrom → runBatch → Loop.Run），DEV HEAD 上 -race 数十次跑 （含把锁删掉的变异）DATA RACE 恒为 0；若将来从车道 worker 触碰这些字段必须加锁 并补 -race 用例。 证据：go test -race -count=3 ./internal/services/channelbridge/ 连跑 6 轮 18/18 全绿、DATA RACE=0；coalesce 子集 -count=80 全绿；整包 -race -count=20 全绿； 全树 go build/vet/test -count=1 全绿。
359 lines
14 KiB
Go
359 lines
14 KiB
Go
package channelbridge
|
||
|
||
// 本文件是**渠道事件留存接线**的验收（design/26 §4/§5；切片 E3）：唯一插入点
|
||
// `recordEvent` 的三条要件各有用例 ——
|
||
//
|
||
// 1. **T1 只计数、不进上下文**（§4.3/§3.1）；
|
||
// 2. **unknown 不进上下文但逐条落行**（§3.7「落库 + 计数 + 绝不进上下文」）；
|
||
// 3. **未接线 / 落库失败时入站照常**（§4.3 的 fail-visible 降级，绝不中断入站）。
|
||
//
|
||
// 另加一道红线钉：普通聊天消息在留存面**零副作用**（主消息路径不受影响）。
|
||
// 两轴映射（类别 → 留存档）与「摘要必须净化、绝不夹带平台原文」另在 events_mapping_test.go。
|
||
//
|
||
// 行/计数的判据本身归 E1（`internal/runtime/events` 的 persistsAsRow 与计数缓冲），
|
||
// 这里刻意用**真的 SQLite 存储**（内存库 + 迁移 0021）做端到端断言，而不是假存储：
|
||
// 否则「T1 不落行」「unknown 落行」会退化成「我们调了 Record」这种同义反复。
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"errors"
|
||
"log/slog"
|
||
"strings"
|
||
"testing"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
)
|
||
|
||
// newTestEventStore 打开一个带迁移 0021 的内存库并构造事件存储（E1 的 SQLStore）。
|
||
// GC 关闭（保留期与间隔都为 0）：本文件验的是写入面，清扫归 E1 自己的用例。
|
||
func newTestEventStore(t *testing.T) *events.SQLStore {
|
||
t.Helper()
|
||
ctx := context.Background()
|
||
conn, err := db.Open(ctx, ":memory:", nil)
|
||
if err != nil {
|
||
t.Fatalf("打开内存库: %v", err)
|
||
}
|
||
t.Cleanup(func() { _ = conn.Close() })
|
||
if err := db.MigrateUp(ctx, conn, nil); err != nil {
|
||
t.Fatalf("内存库跑迁移: %v", err)
|
||
}
|
||
store, err := events.NewStore(events.Config{DB: conn, GCInterval: 0, Retention: 0})
|
||
if err != nil {
|
||
t.Fatalf("events.NewStore: %v", err)
|
||
}
|
||
return store
|
||
}
|
||
|
||
// eventBridgeFixture 组装一条「系统事件入站」所需的桥：受限群会话 + 捕获历史 +
|
||
// 记录调用的循环 + 给定的留存面（nil = 未接线）+ 捕获到 buffer 的 DEBUG 日志
|
||
// （用它断言日志带 source、且降级只记一条 DEBUG）。
|
||
func eventBridgeFixture(t *testing.T, ad adapter.PlatformAdapter, store events.Store) (*Bridge, *stubChatter, *captureHistory, *bytes.Buffer) {
|
||
t.Helper()
|
||
chat := &stubChatter{res: &loop.Result{Text: "嗯"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-g1", IsGroup: true}}
|
||
hist := &captureHistory{}
|
||
var buf bytes.Buffer
|
||
log := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))
|
||
b, err := New(Options{
|
||
Log: log, Loop: chat, Sessions: sess, History: hist,
|
||
Owner: testOwnerProfile("qq-obv11", "1000"), EventStore: store,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
b.Add(ad)
|
||
return b, chat, hist, &buf
|
||
}
|
||
|
||
// eventAdapter 是「按给定归一化结果回答」的最小适配器：它复用 notice_test.go 的
|
||
// noticeAdapter，只多一层固定适配器名（多账号命名口径：`platform:confID`）。
|
||
func eventAdapter(name string, notice adapter.SystemNotice) *noticeAdapter {
|
||
return ¬iceAdapter{fakeAdapter: fakeAdapter{name: name}, notice: notice, ok: true}
|
||
}
|
||
|
||
// failingEventStore 是**一直接着失败**的留存面（库不可用 / 表缺失）：用于钉住
|
||
// 「落库失败只降级，绝不中断入站」（§4.3）。计数不加锁：本文件的调用都在测试
|
||
// goroutine 上同步发生。
|
||
type failingEventStore struct {
|
||
records int
|
||
flushes int
|
||
}
|
||
|
||
func (f *failingEventStore) Record(context.Context, events.Event) error {
|
||
f.records++
|
||
return errors.New("storage unavailable")
|
||
}
|
||
func (f *failingEventStore) Count(context.Context, events.Filter) (int64, error) { return 0, nil }
|
||
func (f *failingEventStore) List(context.Context, events.Filter) ([]events.Event, error) {
|
||
return nil, nil
|
||
}
|
||
func (f *failingEventStore) Counters(context.Context, string) ([]events.Counter, error) {
|
||
return nil, nil
|
||
}
|
||
func (f *failingEventStore) Flush(context.Context) error {
|
||
f.flushes++
|
||
return nil
|
||
}
|
||
func (f *failingEventStore) Sweep(context.Context) (events.SweepStats, error) {
|
||
return events.SweepStats{}, nil
|
||
}
|
||
|
||
// TestRecordEvent_T1只计数不进上下文_正常：心跳是典型的 T1 遥测 —— 它必须**进计数**
|
||
// （「上游到底推没推」的唯一答案），但**绝不**逐条落行（每 ~15s 一帧会把库写胖，§4.3），
|
||
// 也绝不进她的上下文。用例让心跳带着 `SelfRelated=true`（真实心跳帧帧都带 self_id）：
|
||
// 它同时把「帧级遥测不因 self 命中升档」这条取舍钉在这里。
|
||
func TestRecordEvent_T1只计数不进上下文_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestEventStore(t)
|
||
|
||
ad := eventAdapter("qq-obv11:conf0", adapter.SystemNotice{
|
||
Kind: adapter.NoticeOther,
|
||
Category: adapter.NoticeCategoryHeartbeat,
|
||
PlatformType: "heartbeat",
|
||
Ignorable: true,
|
||
SelfRelated: true,
|
||
})
|
||
b, chat, hist, _ := eventBridgeFixture(t, ad, store)
|
||
|
||
b.processFrom(ctx, groupNoticeMessage(), ad)
|
||
|
||
if chat.sessionID != "" {
|
||
t.Fatalf("心跳驱动了 loop：session=%q", chat.sessionID)
|
||
}
|
||
if recs := hist.records(); len(recs) != 0 {
|
||
t.Fatalf("心跳被写进了登记面（会话历史）：%+v", recs)
|
||
}
|
||
rows, err := store.Count(ctx, events.Filter{})
|
||
if err != nil {
|
||
t.Fatalf("Count: %v", err)
|
||
}
|
||
if rows != 0 {
|
||
t.Fatalf("T1 遥测逐条落行 %d 行，want 0（§4.3：心跳逐帧写库是纯浪费）", rows)
|
||
}
|
||
|
||
if err := store.Flush(ctx); err != nil {
|
||
t.Fatalf("Flush: %v", err)
|
||
}
|
||
counters, err := store.Counters(ctx, "qq-obv11")
|
||
if err != nil {
|
||
t.Fatalf("Counters: %v", err)
|
||
}
|
||
if len(counters) != 1 {
|
||
t.Fatalf("计数行数 = %d, want 1（T1 必须进计数）", len(counters))
|
||
}
|
||
got := counters[0]
|
||
if got.Count != 1 || got.Tier != events.TierT1 || got.Category != events.Category(adapter.NoticeCategoryHeartbeat) {
|
||
t.Fatalf("计数 = %+v, want count=1 tier=t1 category=heartbeat", got)
|
||
}
|
||
if got.AccountID != "conf0" || got.PlatformType != "heartbeat" {
|
||
t.Fatalf("计数键 = (account=%q, type=%q), want (conf0, heartbeat)", got.AccountID, got.PlatformType)
|
||
}
|
||
}
|
||
|
||
// TestRecordEvent_unknown不进上下文但落行_正常：未知类型是「兜底不等于放行」的正面用例
|
||
// （§3.7）—— 它照旧逐条落行（可查）并进计数，但**绝不**进她的上下文，且平台原生类型码
|
||
// 必须原样保留（那是「上游到底报了什么类型」的唯一凭据）。
|
||
func TestRecordEvent_unknown不进上下文但落行_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestEventStore(t)
|
||
|
||
ad := eventAdapter("qq-obv11:conf0", adapter.SystemNotice{
|
||
Kind: adapter.NoticeUnknown,
|
||
Category: adapter.NoticeCategoryUnknown,
|
||
PlatformType: "mystery_frame",
|
||
Ignorable: true,
|
||
})
|
||
b, chat, hist, _ := eventBridgeFixture(t, ad, store)
|
||
|
||
b.processFrom(ctx, groupNoticeMessage(), ad)
|
||
|
||
if chat.sessionID != "" {
|
||
t.Fatalf("unknown 事件进了真会话：session=%q", chat.sessionID)
|
||
}
|
||
if recs := hist.records(); len(recs) != 0 {
|
||
t.Fatalf("unknown 事件被登记进会话历史：%+v", recs)
|
||
}
|
||
rows, err := store.List(ctx, events.Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(rows) != 1 {
|
||
t.Fatalf("unknown 事件落行数 = %d, want 1（unknown 必须逐条留痕）", len(rows))
|
||
}
|
||
ev := rows[0]
|
||
if ev.Tier != events.TierT1 || ev.Category != events.CategoryUnknown {
|
||
t.Fatalf("落行档位/类别 = (%s,%s), want (t1,unknown)", ev.Tier, ev.Category)
|
||
}
|
||
if ev.PlatformType != "mystery_frame" {
|
||
t.Fatalf("平台原生类型码 = %q, want mystery_frame（原样保留是排障凭据）", ev.PlatformType)
|
||
}
|
||
if ev.Platform != "qq-obv11" || ev.ChannelID != "g1" || ev.ChannelType != string(adapter.ChannelGroup) {
|
||
t.Fatalf("行来源字段 = (platform=%q, channel=%q, type=%q)", ev.Platform, ev.ChannelID, ev.ChannelType)
|
||
}
|
||
if ev.AccountID != "conf0" {
|
||
t.Fatalf("account_id = %q, want conf0（多账号覆盖；空串留给单账号/未知）", ev.AccountID)
|
||
}
|
||
if !strings.Contains(ev.Summary, "mystery_frame") {
|
||
t.Fatalf("摘要 = %q, want 含未识别的平台原生类型码", ev.Summary)
|
||
}
|
||
}
|
||
|
||
// TestRecordEvent_自身回声升T2落行带出站id_正常 钉住切 M13 修掉的缺口（原 M12 登记）：
|
||
// 她自己发出消息的回声（message_sent）**不是帧级遥测** —— self 命中必须把它从 T1 升到 T2
|
||
// 逐条落行，`message_id`（出站消息 id）随之留在库里；此前它被帧级豁免挡回 T1，id 被丢。
|
||
func TestRecordEvent_自身回声升T2落行带出站id_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestEventStore(t)
|
||
|
||
ad := eventAdapter("qq-obv11:conf0", adapter.SystemNotice{
|
||
Kind: adapter.NoticeOther,
|
||
Category: adapter.NoticeCategoryMessageSent,
|
||
PlatformType: "message_sent",
|
||
Ignorable: true,
|
||
SelfRelated: true,
|
||
RefMessage: "out-msg-42",
|
||
})
|
||
b, chat, hist, _ := eventBridgeFixture(t, ad, store)
|
||
|
||
b.processFrom(ctx, groupNoticeMessage(), ad)
|
||
|
||
if chat.sessionID != "" {
|
||
t.Fatalf("出站回声驱动了 loop：session=%q", chat.sessionID)
|
||
}
|
||
if recs := hist.records(); len(recs) != 0 {
|
||
t.Fatalf("出站回声被登记进会话历史：%+v", recs)
|
||
}
|
||
rows, err := store.List(ctx, events.Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(rows) != 1 {
|
||
t.Fatalf("出站回声落行数 = %d, want 1（self 命中 ⇒ T2 逐条落行）", len(rows))
|
||
}
|
||
ev := rows[0]
|
||
if ev.Tier != events.TierT2 || ev.Category != events.Category(adapter.NoticeCategoryMessageSent) {
|
||
t.Fatalf("落行档位/类别 = (%s,%s), want (t2,message_sent)", ev.Tier, ev.Category)
|
||
}
|
||
if ev.MessageID != "out-msg-42" {
|
||
t.Fatalf("出站消息 id = %q, want out-msg-42（id 必须留在库里）", ev.MessageID)
|
||
}
|
||
}
|
||
|
||
// TestRecordEvent_未接线时入站照常_异常：`Options.EventStore == nil` 时降级为
|
||
// 「内存计数 + 首次一条 DEBUG」，**绝不**中断入站 —— 高价值事件的升级真会话行为必须
|
||
// 逐字不变（fail-visible 而非 fail-crash，§4.3）。
|
||
func TestRecordEvent_未接线时入站照常_异常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
|
||
ad := eventAdapter("qq-obv11", adapter.SystemNotice{
|
||
Kind: adapter.NoticeLeave,
|
||
Category: adapter.NoticeCategoryLeave,
|
||
PlatformType: "group_decrease",
|
||
SafeText: "系统通知：她（机器人账号）被移出了群聊",
|
||
HighValue: true,
|
||
SelfRelated: true,
|
||
})
|
||
b, chat, _, buf := eventBridgeFixture(t, ad, nil)
|
||
|
||
b.processFrom(ctx, groupNoticeMessage(), ad)
|
||
|
||
if chat.sessionID != "sess-g1" {
|
||
t.Fatalf("未接线改变了入站行为：高价值事件没进真会话（session=%q）", chat.sessionID)
|
||
}
|
||
stats := b.EventFallback()
|
||
if stats.Frames != 1 {
|
||
t.Fatalf("降级计数 = %d, want 1（未接线的帧也必须留下可查的计数）", stats.Frames)
|
||
}
|
||
if stats.Keys["qq-obv11/group_decrease"] != 1 {
|
||
t.Fatalf("降级计数键 = %v, want qq-obv11/group_decrease=1", stats.Keys)
|
||
}
|
||
logged := buf.String()
|
||
if !strings.Contains(logged, "source=services/channelbridge/events") {
|
||
t.Fatalf("降级日志没有来源标记（AGENTS §10 禁无来源裸日志）：%q", logged)
|
||
}
|
||
if !strings.Contains(logged, "channel event not recorded") {
|
||
t.Fatalf("未接线时没有留下可见痕迹：%q", logged)
|
||
}
|
||
|
||
// 「DEBUG 一次」：第二帧不再刷日志，但计数继续累加（有界、不刷屏）。
|
||
buf.Reset()
|
||
b.processFrom(ctx, groupNoticeMessage(), ad)
|
||
if again := b.EventFallback().Frames; again != 2 {
|
||
t.Fatalf("降级计数 = %d, want 2", again)
|
||
}
|
||
if got := buf.String(); strings.Contains(got, "channel event not recorded") {
|
||
t.Fatalf("降级日志被逐帧刷写（应只在首帧记一条 DEBUG）：%q", got)
|
||
}
|
||
}
|
||
|
||
// TestRecordEvent_落库失败仍不中断入站_异常：存储报错 ⇒ WARN + 内存计数 + 继续 ——
|
||
// **不关留存面、不中断回合、不让入站失败**（§4.3）。
|
||
func TestRecordEvent_落库失败仍不中断入站_异常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := &failingEventStore{}
|
||
|
||
ad := eventAdapter("qq-obv11", adapter.SystemNotice{
|
||
Kind: adapter.NoticeLeave,
|
||
Category: adapter.NoticeCategoryLeave,
|
||
PlatformType: "group_decrease",
|
||
SafeText: "系统通知：她（机器人账号）被移出了群聊",
|
||
HighValue: true,
|
||
})
|
||
b, chat, _, buf := eventBridgeFixture(t, ad, store)
|
||
|
||
b.processFrom(ctx, groupNoticeMessage(), ad)
|
||
|
||
if store.records != 1 {
|
||
t.Fatalf("留存面被调用 %d 次, want 1", store.records)
|
||
}
|
||
if chat.sessionID != "sess-g1" {
|
||
t.Fatalf("落库失败中断了入站（高价值事件没进真会话）：session=%q", chat.sessionID)
|
||
}
|
||
if stats := b.EventFallback(); stats.Frames != 1 {
|
||
t.Fatalf("落库失败的帧没有进降级计数：%+v", stats)
|
||
}
|
||
logged := buf.String()
|
||
if !strings.Contains(logged, "level=WARN") ||
|
||
!strings.Contains(logged, "channel event not persisted") ||
|
||
!strings.Contains(logged, "source=services/channelbridge/events") {
|
||
t.Fatalf("落库失败的可观测面不足（要 WARN + source）：%q", logged)
|
||
}
|
||
}
|
||
|
||
// TestRecordEvent_聊天消息零副作用_正常：普通聊天消息不是渠道事件 —— 留存面**一次都
|
||
// 不该被调用**（主消息路径不受影响，§5.1）。
|
||
func TestRecordEvent_聊天消息零副作用_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestEventStore(t)
|
||
|
||
ad := eventAdapter("qq-obv11", adapter.SystemNotice{Kind: adapter.NoticeJoin})
|
||
b, _, _, _ := eventBridgeFixture(t, ad, store)
|
||
|
||
msg := groupNoticeMessage()
|
||
msg.ContentType, msg.Content = adapter.ContentText, "大家好"
|
||
msg.Sender.Role, msg.SenderID = adapter.RoleMember, "42"
|
||
b.processFrom(ctx, msg, ad)
|
||
|
||
rows, err := store.Count(ctx, events.Filter{})
|
||
if err != nil {
|
||
t.Fatalf("Count: %v", err)
|
||
}
|
||
if rows != 0 || store.PendingKeys() != 0 {
|
||
t.Fatalf("普通聊天消息进了留存面：rows=%d pending=%d", rows, store.PendingKeys())
|
||
}
|
||
if stats := b.EventFallback(); stats.Frames != 0 {
|
||
t.Fatalf("普通聊天消息进了降级计数：%+v", stats)
|
||
}
|
||
}
|