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 已有结论），「留存无条件、在任何早退之前」这条性质不变
This commit is contained in:
AskaEth committed
1 parent
7aaf7731dc
commit
afe9afb1b5
12 files changed
+1689
-2
No files matched your search
@@ -0,0 +1,188 @@
|
||||
package main
|
||||
|
||||
// 本文件把**渠道事件留存**接进组合根（design/26 §4.3/§4.4；切片 E3+E4）：按配置构造
|
||||
// 事件存储、交给桥接层注入（wire_channels.go 的 channelbridge.Options.EventStore），
|
||||
// 拉起常驻的「保留期清扫 + 内存计数落盘」循环，并在退出前最后一次落盘。
|
||||
//
|
||||
// 三条取舍逐字对齐参照实现 `wire_document_gc.go`（design/26 §4.4 明写「不另造」）：
|
||||
//
|
||||
// 1. **启动期清扫不阻塞启动**：它是常驻 goroutine 的第一件事（紧随子系统启动之后
|
||||
// 开始，网关监听之前就绪），而不是在 newApp/run 里同步等它跑完。
|
||||
// 2. **失败不退出进程**：清扫 / 落盘失败只记一条 WARN 并等下一轮；一轮失败只是
|
||||
// 「这一轮没扫干净」，绝不让进程退出。
|
||||
// 3. **关闭 = 显式配 ≤0**：`CHANNEL_EVENT_RETENTION=0`（或负值）或
|
||||
// `CHANNEL_EVENT_GC_INTERVAL=0` 都不开启清扫，NewStore 会把关闭原因记进日志
|
||||
// （静默的「配了却不生效」不存在）。
|
||||
//
|
||||
// ⚠️ 与参照实现的一处**已知差异**（如实登记，不假装对齐）：docGC 的 `stop()` 由
|
||||
// `wire_services.go` 的 close 调用，而该文件由别的切片在改（本片不许动）。因此这里的
|
||||
// 退出路径做成**循环自己的 ctx.Done 分支**：ctx 取消即用独立超时的 context 落最后一次盘。
|
||||
// 语义与「退出前最后一次 Flush」相同，区别只在**不阻塞** close —— 计数本就是尽力而为的
|
||||
// 近似值（§4.3），并与关库并发（真撞上只记 WARN，不丢留存面本身）。
|
||||
//
|
||||
// 三个键的读取走 `config.LoadDurationEnv`（进程环境 > `.env` > E1 导出的默认常量）：
|
||||
// `.env` 不写回进程环境，直接 os.Getenv 会让只写在 `.env` 里的声明静默失效。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
|
||||
)
|
||||
|
||||
// 三个 env 键名：必须与 `.env.example` **逐字一致**（AGENTS §10 的 E6 条目）。
|
||||
const (
|
||||
channelEventRetentionEnv = "CHANNEL_EVENT_RETENTION"
|
||||
channelEventGCIntervalEnv = "CHANNEL_EVENT_GC_INTERVAL"
|
||||
channelEventCounterFlushEnv = "CHANNEL_EVENT_COUNTER_FLUSH"
|
||||
)
|
||||
|
||||
// channelEventFlushTimeout 是**退出前最后一次 Flush** 的时长上界。它必须配一个独立的
|
||||
// context：ctx 在这一刻已被取消，拿它去写库只会立刻拿到 context.Canceled —— 那正是
|
||||
// 「退出时最后一次 flush」最容易悄悄失效的地方。
|
||||
const channelEventFlushTimeout = 5 * time.Second
|
||||
|
||||
// wireChannelEvents 构造事件留存面并挂到 app 上（随后由 wireChannels 注入桥接层）。
|
||||
// 它**不**启动任何 goroutine：常驻循环由 main 的 startChannelEvents 在子系统启动之后
|
||||
// 拉起（与文档 GC 的分工逐字相同）。
|
||||
func (a *app) wireChannelEvents() {
|
||||
a.channelEvents = a.newChannelEventStore(envPath)
|
||||
}
|
||||
|
||||
// newChannelEventStore 按配置构造事件存储。库未接（`a.db == nil`，只可能出现在测试
|
||||
// 夹具里）时返回 nil：桥接层随即走「未接线」的 fail-visible 降级路径（内存计数 +
|
||||
// 首次一条 DEBUG），入站流水线照常，绝不因为没有存储而失败或丢帧。
|
||||
//
|
||||
// 默认值一律取 E1（`internal/runtime/events`）导出的常量 —— **不在这里另写一份字面量**：
|
||||
// 两处各写一份 720h/1h/30s，改一处就会让「文档承诺的默认值」与「代码里的默认值」漂移。
|
||||
// `MaxPendingKeys` 没有 env 键（design/26 §4.4 只给了三个旋钮），同样取 E1 的默认常量。
|
||||
func (a *app) newChannelEventStore(envFilePath string) *events.SQLStore {
|
||||
if a.db == nil {
|
||||
a.log.Warn("channel event store skipped: no database connection")
|
||||
return nil
|
||||
}
|
||||
store, err := events.NewStore(events.Config{
|
||||
DB: a.db,
|
||||
Log: a.log,
|
||||
FlushInterval: durationEnv(a.log, envFilePath, channelEventCounterFlushEnv, events.DefaultFlushInterval),
|
||||
MaxPendingKeys: events.DefaultMaxPendingKeys,
|
||||
GCInterval: durationEnv(a.log, envFilePath, channelEventGCIntervalEnv, events.DefaultGCInterval),
|
||||
Retention: durationEnv(a.log, envFilePath, channelEventRetentionEnv, events.DefaultRetention),
|
||||
})
|
||||
if err != nil {
|
||||
// NewStore 唯一的失败是 DB 为 nil（上面已挡掉）：仍如实处理，绝不吞错。
|
||||
a.log.Warn("channel event store unavailable", "err", err)
|
||||
return nil
|
||||
}
|
||||
return store
|
||||
}
|
||||
|
||||
// durationEnv 读一个 duration 型 env 键（进程环境 > `.env` > def）。配了但读不懂时记
|
||||
// 一条带原始取值的 WARN 并回落 def —— 错的是那一个键，不该让整条留存面失效
|
||||
// （AGENTS §10：非法配置不许静默失效，但也不许一票否决）。
|
||||
func durationEnv(log *slog.Logger, envFilePath, key string, def time.Duration) time.Duration {
|
||||
value, err := config.LoadDurationEnv(envFilePath, key, def)
|
||||
if err != nil {
|
||||
log.Warn("channel event env ignored, falling back to default",
|
||||
"env", key, "error", err, "fallback", def.String())
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
// channelEventStore 返回注入桥接层的留存面：未接线时返回**真正的 nil 接口**，而不是
|
||||
// 「装着 nil 指针的接口」—— 后者在 Go 里不等于 nil，会让桥接层的「未接线 ⇒ 内存计数」
|
||||
// 判据失效（症状是每帧都撞一次 nil 指针 panic）。
|
||||
func (a *app) channelEventStore() events.Store {
|
||||
if a.channelEvents == nil {
|
||||
return nil
|
||||
}
|
||||
return a.channelEvents
|
||||
}
|
||||
|
||||
// startChannelEvents 拉起留存维护循环（未接线时是空操作）。
|
||||
func (a *app) startChannelEvents(ctx context.Context) {
|
||||
if a.channelEvents == nil {
|
||||
a.log.Info("channel event upkeep disabled: store not wired",
|
||||
"effect", "events are counted in memory only (bridged to the api layer counters)")
|
||||
return
|
||||
}
|
||||
keeper := &channelEventKeeper{
|
||||
store: a.channelEvents,
|
||||
log: logger.WithSource(a.log, "cmd/cyrene/wire_channel_events"),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
a.channelEventKeeper = keeper
|
||||
go keeper.loop(ctx)
|
||||
}
|
||||
|
||||
// channelEventKeeper 是渠道事件留存的常驻维护任务：保留期清扫 + 内存计数落盘 +
|
||||
// 退出前最后一次落盘。它持有**存储句柄**（不是 app），因此生命周期与用法一目了然。
|
||||
type channelEventKeeper struct {
|
||||
store *events.SQLStore
|
||||
log *slog.Logger
|
||||
// done 在循环退出时关闭。它让「退出前最后一次 Flush」**可被观测**：测试据此确认
|
||||
// 内存里的计数真的落了盘，而不是随 goroutine 一起消失。
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// loop 是维护循环：启动期先清扫一次、落一次盘，此后按各自间隔定时跑，直到 ctx 被取消
|
||||
// （取消时落最后一次盘再退出）。
|
||||
func (k *channelEventKeeper) loop(ctx context.Context) {
|
||||
defer close(k.done)
|
||||
k.sweep(ctx)
|
||||
k.flush(ctx)
|
||||
|
||||
gcTicks, stopGC := k.gcTicker()
|
||||
defer stopGC()
|
||||
flushTicks := time.NewTicker(k.store.FlushInterval())
|
||||
defer flushTicks.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
k.flushOnShutdown()
|
||||
return
|
||||
case <-gcTicks:
|
||||
k.sweep(ctx)
|
||||
case <-flushTicks.C:
|
||||
k.flush(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// gcTicker 给出清扫的定时通道；GC 关闭时返回 **nil 通道**（select 上永不触发）。
|
||||
// `time.NewTicker(0)` 会 panic，所以「配了 ≤0 = 关闭」必须在**这里**就分岔，而不是
|
||||
// 建一个零间隔 ticker 再指望它不跑。
|
||||
func (k *channelEventKeeper) gcTicker() (<-chan time.Time, func()) {
|
||||
if !k.store.Enabled() {
|
||||
return nil, func() {}
|
||||
}
|
||||
ticker := time.NewTicker(k.store.Interval())
|
||||
return ticker.C, ticker.Stop
|
||||
}
|
||||
|
||||
// sweep 跑一轮保留期清扫。每轮的**成功**汇总由 store 自己记（E1 的 Sweep 已记一条，
|
||||
// 含删除行数与批次），这里只负责把「这一轮整个没跑成」记成 WARN 并让循环继续。
|
||||
func (k *channelEventKeeper) sweep(ctx context.Context) {
|
||||
if _, err := k.store.Sweep(ctx); err != nil {
|
||||
k.log.Warn("channel event sweep failed", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// flush 把内存里待落盘的计数写进库；失败只 WARN（计数是尽力而为的近似值，§4.3）。
|
||||
func (k *channelEventKeeper) flush(ctx context.Context) {
|
||||
if err := k.store.Flush(ctx); err != nil {
|
||||
k.log.Warn("channel event counter flush failed", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// flushOnShutdown 是退出前的最后一次落盘：用独立超时的 context（此刻调用方的 ctx
|
||||
// 已取消），失败只 WARN —— 退出路径绝不因为一次写库失败而 panic 或阻塞。
|
||||
func (k *channelEventKeeper) flushOnShutdown() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), channelEventFlushTimeout)
|
||||
defer cancel()
|
||||
k.flush(ctx)
|
||||
}
|
||||
Reference in new issue
Block a user