Files
Cyrene_v2/cmd/cyrene/wire_channel_events_test.go
AskaEth afe9afb1b5 feat(events): 渠道事件桥接接线 + 组合根(design/26 E3+E4)
- E3:channelbridge/events.go 唯一插入点 recordEvent(两轴映射 Category→Tier、摘要恒过 SanitizeNoticeText、角色/账号只认白名单)+ Options.EventStore 注入
  - 未接线 store=nil ⇒ 内存计数 + 首次 DEBUG,绝不中断入站(变异实测:改成直接调 Record ⇒ panic,测试红)
  - 普通聊天消息零副作用(变异:去掉 Applicable 早退 ⇒ rows=1 红)
  - T1 遥测只计数不落行(变异:心跳误判 T2 ⇒ 「逐条落行 1 行,want 0」红)
- E4:wire_channel_events.go(启动期清扫 + 双 ticker + 退出 flush,逐字对齐 wire_document_gc 的三条取舍)、wire_channels 注入、main.go 加一行 startChannelEvents
- internal/config/duration_env.go:新 env 键的读取口径(进程环境 > .env > 默认;「没写」与「写了空」都当没配;非法值回落 + 报错)——避免「只写在 .env 里被静默忽略」
- 三个新 env 键:CHANNEL_EVENT_RETENTION / GC_INTERVAL / COUNTER_FLUSH(模板同批更新,测试钉住键名一致)
- 插入点比设计示意图晚一步(放 preEntryReview 之后,只归一化一次且 NoticeClass 已有结论),「留存无条件、在任何早退之前」这条性质不变
2026-10-05 14:10:24 +08:00

342 lines
14 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package main
// 本文件是**渠道事件留存组合根接线**的验收(design/26 §4.3/§4.4;切片 E3+E4)。
//
// 两道并行的取证:
//
// 1. **接线存在性**(与 TestWireDocumentGC_IsWired 同一手法):组合根本身无法在单元
// 测试里实例化(开库、迁移、绑端口,而后端还在运行),因此「存储真的被构造」「真的
// 注入桥接层」「常驻循环真的被拉起」「退出前真的落最后一次盘」只能在源码层钉住 ——
// 少任何一处,留存都会**静默不工作**(事件照丢、日志里一句都没有),而这不
// 会让任何既有用例变红。最后一段还核对 `.env.example` 的三个键(AGENTS §10)。
// 2. **行为**:三个 env 键的默认值必须来自 E1 导出的常量(**不另写一份字面量**)、
// 显式 ≤0 = 关闭清扫、只写在 `.env` 里也真实生效、非法取值回落默认 + WARN;
// 以及「退出前最后一次落盘」真的把内存里的计数写进库。
import (
"bytes"
"context"
"log/slog"
"os"
"path/filepath"
"strings"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
)
// TestWireChannelEvents_IsWired 钉住组合根的四个接缝 + `.env.example` 同步。
func TestWireChannelEvents_IsWired(t *testing.T) {
t.Parallel()
seams := map[string][]string{
"wire_channel_events.go": {
// 存储真的被构造,且默认值取 E1 导出的常量(不另写字面量)。
"events.NewStore(events.Config{",
"MaxPendingKeys: events.DefaultMaxPendingKeys",
"durationEnv(a.log, envFilePath, channelEventCounterFlushEnv, events.DefaultFlushInterval)",
"durationEnv(a.log, envFilePath, channelEventGCIntervalEnv, events.DefaultGCInterval)",
"durationEnv(a.log, envFilePath, channelEventRetentionEnv, events.DefaultRetention)",
// 未接线时交出**真 nil 接口**(装 nil 指针的接口不等于 nil,会让桥接层的
// 「未接线 ⇒ 内存计数」降级判据失效)。
"func (a *app) channelEventStore() events.Store {",
// GC 关闭时不许建零间隔 ticker(time.NewTicker(0) 会 panic)。
"if !k.store.Enabled() {",
// 退出前最后一次落盘,用独立 ctx(此刻调用方的 ctx 已取消)。
"k.flushOnShutdown()",
},
"wire_channels.go": {
"EventStore: a.channelEventStore(),",
},
"wire.go": {
// 必须在 wireChannels 之前构造:桥接层在构造时就注入它。
"a.wireChannelEvents()",
},
"main.go": {
// 常驻循环:没有这一行,留存面只存在于代码里而从不维护(清扫与计数落盘都不跑)。
"a.startChannelEvents(ctx)",
},
}
for file, wants := range seams {
src, err := os.ReadFile(file)
if err != nil {
t.Fatalf("read %s: %v", file, err)
}
for _, want := range wants {
if !strings.Contains(string(src), want) {
t.Fatalf("%s 里找不到 %q —— 渠道事件留存没有被接上组合根", file, want)
}
}
}
// 三个新键必须与代码里的键名**逐字一致**(AGENTS §10 的 E6 条目)。
src, err := os.ReadFile("wire_channel_events.go")
if err != nil {
t.Fatalf("read wire_channel_events.go: %v", err)
}
raw, err := os.ReadFile(filepath.Join("..", "..", ".env.example"))
if err != nil {
t.Fatalf("read .env.example: %v", err)
}
for _, key := range []string{channelEventRetentionEnv, channelEventGCIntervalEnv, channelEventCounterFlushEnv} {
if !strings.Contains(string(src), `"`+key+`"`) {
t.Fatalf("wire_channel_events.go 里没有键名常量 %q", key)
}
if !strings.Contains(string(raw), key+"=") {
t.Fatalf(".env.example 里找不到 %q(新 env 键必须同步模板,AGENTS §10)", key+"=")
}
}
}
// channelEventTestApp 构造一个只带留存面所需依赖的 app(库 + 静默 logger)。
func channelEventTestApp(t *testing.T, log *slog.Logger) *app {
t.Helper()
if log == nil {
log = testLog()
}
return &app{db: openMigratedTestDB(t, filepath.Join(t.TempDir(), "cyrene.db")), log: log}
}
// unsetChannelEventEnv 把三个键从**进程环境**里拿掉并在用例结束后复原。
//
// ⚠️ 不能用 `t.Setenv(key, "")` 代替:空值在优先级链里仍然「存在」,会**盖住** `.env`
// 里的同名声明(lookupRaw 先看 os.LookupEnv),而「只写在 .env 里也生效」正是要验的东西。
func unsetChannelEventEnv(t *testing.T) {
t.Helper()
for _, key := range []string{channelEventRetentionEnv, channelEventGCIntervalEnv, channelEventCounterFlushEnv} {
key := key
prev, had := os.LookupEnv(key)
if had {
t.Cleanup(func() { _ = os.Setenv(key, prev) })
} else {
t.Cleanup(func() { _ = os.Unsetenv(key) })
}
if err := os.Unsetenv(key); err != nil {
t.Fatalf("unset %s: %v", key, err)
}
}
}
// TestNewChannelEventStore_默认值取E1导出常量_正常:未配置时三个旋钮必须**就是**
// E1 导出的默认值(720h / 1h / 30s),并且清扫是开启的 —— 若这里另写了一份字面量,
// 「文档承诺的默认值」与「代码里的默认值」迟早漂移。
func TestNewChannelEventStore_默认值取E1导出常量_正常(t *testing.T) {
unsetChannelEventEnv(t)
store := channelEventTestApp(t, nil).newChannelEventStore("")
if store == nil {
t.Fatal("有库时必须构造出事件存储")
}
if store.Retention() != events.DefaultRetention {
t.Fatalf("Retention = %v, want %v(E1 导出的默认常量)", store.Retention(), events.DefaultRetention)
}
if store.Interval() != events.DefaultGCInterval {
t.Fatalf("GC 间隔 = %v, want %v", store.Interval(), events.DefaultGCInterval)
}
if store.FlushInterval() != events.DefaultFlushInterval {
t.Fatalf("计数落盘间隔 = %v, want %v", store.FlushInterval(), events.DefaultFlushInterval)
}
if !store.Enabled() {
t.Fatal("未配置时清扫必须开启(漏配 != 关闭)")
}
}
// TestNewChannelEventStore_显式零关闭清扫_边界:`≤0 = 关闭`(design/26 §4.4)——
// 0 刻意**不**解释成「立即清扫」,那会让一次笔误抹掉唯一的取证面。
func TestNewChannelEventStore_显式零关闭清扫_边界(t *testing.T) {
cases := []struct {
name string
key string
value string
}{
{"保留期为 0", channelEventRetentionEnv, "0"},
{"保留期为负", channelEventRetentionEnv, "-1h"},
{"清扫间隔为 0", channelEventGCIntervalEnv, "0"},
}
for _, tc := range cases {
tc := tc
t.Run(tc.name, func(t *testing.T) {
unsetChannelEventEnv(t)
t.Setenv(tc.key, tc.value)
store := channelEventTestApp(t, nil).newChannelEventStore("")
if store == nil {
t.Fatal("配置非法不该让存储整个消失")
}
if store.Enabled() {
t.Fatalf("%s=%s 时清扫仍开启,want 关闭(显式 ≤0 = 关闭)", tc.key, tc.value)
}
})
}
}
// TestNewChannelEventStore_只写在env文件里也生效_正常:`.env` 不写回进程环境,直接
// os.Getenv 会**静默忽略**它(= 配了却不生效)。本用例把三个键只写进临时 `.env`,
// 取值必须原样生效。
func TestNewChannelEventStore_只写在env文件里也生效_正常(t *testing.T) {
unsetChannelEventEnv(t)
path := filepath.Join(t.TempDir(), ".env")
body := strings.Join([]string{
channelEventRetentionEnv + "=24h",
channelEventGCIntervalEnv + "=30m",
channelEventCounterFlushEnv + "=5s",
}, "\n") + "\n"
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatalf("写临时 .env: %v", err)
}
store := channelEventTestApp(t, nil).newChannelEventStore(path)
if store == nil {
t.Fatal("有库时必须构造出事件存储")
}
if store.Retention() != 24*time.Hour || store.Interval() != 30*time.Minute || store.FlushInterval() != 5*time.Second {
t.Fatalf("取值 = (%v, %v, %v), want (24h, 30m, 5s)",
store.Retention(), store.Interval(), store.FlushInterval())
}
}
// TestNewChannelEventStore_非法取值回落默认并告警_异常:配了但读不懂 ⇒ 回落 E1 默认值
// **且**记一条带键名与原始取值的 WARN(AGENTS §10:非法配置不许静默失效,也不许一票
// 否决整条留存面)。
func TestNewChannelEventStore_非法取值回落默认并告警_异常(t *testing.T) {
unsetChannelEventEnv(t)
path := filepath.Join(t.TempDir(), ".env")
if err := os.WriteFile(path, []byte(channelEventRetentionEnv+"=三十天\n"), 0o600); err != nil {
t.Fatalf("写临时 .env: %v", err)
}
var buf bytes.Buffer
log := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))
store := channelEventTestApp(t, log).newChannelEventStore(path)
if store == nil {
t.Fatal("单个键非法不该让存储整个消失")
}
if store.Retention() != events.DefaultRetention {
t.Fatalf("非法取值时的保留期 = %v, want %v", store.Retention(), events.DefaultRetention)
}
logged := buf.String()
if !strings.Contains(logged, "level=WARN") || !strings.Contains(logged, channelEventRetentionEnv) {
t.Fatalf("非法取值没有留下带键名的 WARN:%q", logged)
}
}
// TestChannelEventStore_未接线时交出真nil接口_边界:未接数据库(或构造失败)时
// `channelEventStore()` 必须返回**真 nil 接口** —— 返回「装着 nil 指针的接口」会让
// 桥接层的 nil 判据失效,症状是每帧撞一次 nil 指针 panic,而单测全绿。
func TestChannelEventStore_未接线时交出真nil接口_边界(t *testing.T) {
blank := &app{log: testLog()}
if store := blank.newChannelEventStore(""); store != nil {
t.Fatalf("库未接时不该构造存储:%+v", store)
}
var asInterface events.Store = blank.channelEventStore()
if asInterface != nil {
t.Fatalf("未接线时必须交出真 nil 接口,拿到的是 %#v", asInterface)
}
}
// TestChannelEventKeeper_退出前最后一次落盘_正常:ctx 取消时循环必须用**独立**的
// context 把内存里的计数写进库。这里把已取消的 ctx 直接交给 loop(同步跑完)——
// 若 `flushOnShutdown` 用调用方那个已取消的 ctx 落盘,它会立刻拿到 context.Canceled,
// 待落盘的计数就会留在内存里(用例据此变红)。
func TestChannelEventKeeper_退出前最后一次落盘_正常(t *testing.T) {
unsetChannelEventEnv(t)
a := channelEventTestApp(t, nil)
store := a.newChannelEventStore("")
if store == nil {
t.Fatal("有库时必须构造出事件存储")
}
ctx := context.Background()
// 一条只计数的 T1 事件:它在内存里待落盘(逐条落行的那类会立刻写库,掩盖问题)。
if err := store.Record(ctx, events.Event{
Platform: events.PlatformQQ,
Tier: events.TierT1,
Category: events.CategoryHeartbeat,
PlatformType: "heartbeat",
}); err != nil {
t.Fatalf("Record: %v", err)
}
if store.PendingKeys() == 0 {
t.Fatal("计数没有留在内存里,用例前提不成立")
}
cancelled, cancel := context.WithCancel(ctx)
cancel()
keeper := &channelEventKeeper{store: store, log: testLog(), done: make(chan struct{})}
keeper.loop(cancelled)
if store.PendingKeys() != 0 {
t.Fatalf("退出时仍有 %d 个键待落盘:最后一次 Flush 没跑", store.PendingKeys())
}
counters, err := store.Counters(ctx, events.PlatformQQ)
if err != nil {
t.Fatalf("Counters: %v", err)
}
if len(counters) != 1 || counters[0].Count != 1 {
t.Fatalf("退出后库里的计数 = %+v, want 一条 count=1 的心跳", counters)
}
select {
case <-keeper.done:
default:
t.Fatal("loop 退出时没有关闭 done(退出路径不可观测)")
}
}
// TestChannelEventKeeper_GC关闭时不建零间隔ticker_边界:`time.NewTicker(0)` 会 panic,
// 所以「配了 ≤0 = 关闭清扫」必须在建 ticker 之前分岔(关闭 ⇒ nil 通道,select 上永不触发)。
func TestChannelEventKeeper_GC关闭时不建零间隔ticker_边界(t *testing.T) {
t.Run("关闭时不建 ticker", func(t *testing.T) {
unsetChannelEventEnv(t)
t.Setenv(channelEventGCIntervalEnv, "0")
store := channelEventTestApp(t, nil).newChannelEventStore("")
if store == nil || store.Enabled() {
t.Fatalf("GC 应关闭:store=%v enabled=%v", store, store != nil && store.Enabled())
}
ticks, stop := (&channelEventKeeper{store: store, log: testLog()}).gcTicker()
if ticks != nil {
t.Fatal("GC 关闭时不该建 ticker(零间隔 ticker 会 panic)")
}
stop()
})
t.Run("开启时给出定时通道", func(t *testing.T) {
unsetChannelEventEnv(t)
store := channelEventTestApp(t, nil).newChannelEventStore("")
if store == nil || !store.Enabled() {
t.Fatal("默认配置下 GC 应开启")
}
ticks, stop := (&channelEventKeeper{store: store, log: testLog()}).gcTicker()
if ticks == nil {
t.Fatal("GC 开启时必须给出定时通道")
}
stop()
})
}
// TestChannelEventEnv_模板取值是合法duration_正常 是 E6(协同清单)自查的机械化版本:
// 三个键名必须能在 `.env.example` 里逐字找到,且模板值本身就是合法 duration(占位符形态,
// 从不放凭据)—— 否则照模板部署的新实例会得到一条 WARN 并**静默**回落到默认值。
func TestChannelEventEnv_模板取值是合法duration_正常(t *testing.T) {
t.Parallel()
raw, err := os.ReadFile(filepath.Join("..", "..", ".env.example"))
if err != nil {
t.Fatalf("read .env.example: %v", err)
}
lines := strings.Split(string(raw), "\n")
for _, key := range []string{channelEventRetentionEnv, channelEventGCIntervalEnv, channelEventCounterFlushEnv} {
found := false
for _, line := range lines {
value, ok := strings.CutPrefix(line, key+"=")
if !ok {
continue
}
found = true
if _, err := time.ParseDuration(value); err != nil {
t.Fatalf("%s 的模板值 %q 不是合法 duration", key, value)
}
}
if !found {
t.Fatalf(".env.example 缺少 %s(AGENTS §10:键名必须与代码逐字一致)", key)
}
}
}