feat(events): 渠道事件存储地基（Event/Store + SQLite 实现 + 有界计数聚合 + 保留期 GC + 迁移 0021）
- 新包 internal/runtime/events：Tier/Category/Platform/ChannelType/ActorRole 具名常量 + Event/Filter/Counter + Store 六方法 - 留存口径：T2/T3 逐条落行；T1 只计数；unknown / decode_error / link 三例外逐条落行（链路是否活着是事故定案唯一证据） - 计数：内存聚合 + 上限 512 的有界缓冲 + 满即整批 upsert，落盘失败回填而不丢；GC：保留期 720h/间隔 1h，≤0 = 关闭 - 迁移 0021_channel_events（两表 + 6 索引，含计数 upsert 唯一键）；Event 无 flag/正文/URL 字段（design/26 §4.2 红线）
This commit is contained in:
AskaEth committed
1 parent
52dac4575f
commit
ec3f6e810a
10 files changed
+2172
No files matched your search
@@ -0,0 +1,261 @@
|
||||
package events
|
||||
|
||||
// 本文件归属**计数面**（design/26 §4.1 表二 + §4.3；切片 E1）：内存聚合 + 有界
|
||||
// 缓冲 + 定期 flush + 逐类汇总读取。
|
||||
//
|
||||
// 为什么计数不逐帧落行：心跳 / 在线状态是**高频遥测**，逐帧插一行会把库写胖，而
|
||||
// 问题「上游到底推过没有」只要求「某个平台某类事件推过几次」。按
|
||||
// `(platform, account_id, platform_type)` upsert 之后，行数上界是「平台数 × 事件
|
||||
// 类型数」，与帧数无关（§4.1 表二）。
|
||||
//
|
||||
// 三条不变量（每条都有用例钉住）：
|
||||
//
|
||||
// 1. **有界**：待落盘的键数绝不超过 `Config.MaxPendingKeys`。缓冲满了先整批落盘，
|
||||
// 绝不无界增长（§4.3 的背压；对齐 AGENTS §10 的压测纪律：有界，不靠堆进程）。
|
||||
// 2. **尽力而为**：落盘失败只记 WARN / 丢键记 DEBUG，**绝不**让 `Record` 失败 ——
|
||||
// 入站流水线不该因为计数写不进去而失败或丢帧（§4.3）。
|
||||
// 3. **首次权威**：同一个键的 tier/category 只认第一次（upsert 不刷写这两列），
|
||||
// 与 `documents` 的 `ON CONFLICT(name) DO NOTHING` 同口径。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// counterKey 是计数聚合的键，**逐列对齐**迁移 0021 的唯一索引
|
||||
// `idx_channel_event_counters_key (platform, account_id, platform_type)`。
|
||||
// 键里刻意不含 tier/category：它们由「平台原生类型」决定，不该让同一类帧分裂成
|
||||
// 两行（那会让计数上界失控，也让「推过几次」的答案分叉）。
|
||||
type counterKey struct {
|
||||
platform string
|
||||
accountID string
|
||||
platformType string
|
||||
}
|
||||
|
||||
// counterEntry 是一个键的累计值与归类。
|
||||
type counterEntry struct {
|
||||
tier Tier
|
||||
category Category
|
||||
count int64
|
||||
firstSeen time.Time
|
||||
lastSeen time.Time
|
||||
}
|
||||
|
||||
// counterBuffer 是内存计数聚合。它由互斥量保护（Record 可能来自多条渠道 goroutine）。
|
||||
//
|
||||
// 锁的纪律（AGENTS §六）：**锁内绝不做 IO** —— drain 在锁内只做「搬运 + 清空」，
|
||||
// 真正的落盘在锁外进行；失败时的回填也走一次短暂的加锁。
|
||||
type counterBuffer struct {
|
||||
mu sync.Mutex
|
||||
max int
|
||||
pending map[counterKey]counterEntry
|
||||
}
|
||||
|
||||
// newCounterBuffer 构造聚合缓冲；最大键数 ≤0 时取 DefaultMaxPendingKeys。
|
||||
func newCounterBuffer(max int) *counterBuffer {
|
||||
if max <= 0 {
|
||||
max = DefaultMaxPendingKeys
|
||||
}
|
||||
return &counterBuffer{max: max, pending: make(map[counterKey]counterEntry, max)}
|
||||
}
|
||||
|
||||
// add 把一帧的计数累加进缓冲。返回值报告**这次计数是否被吞掉**：缓冲已达最大键数
|
||||
// 且这是一个新键时为 false（该键不落内存）。它是「有界」在代码里的落点 ——
|
||||
// 无论落盘成功与否，`len()` 都绝不超过 `max`。
|
||||
func (b *counterBuffer) add(key counterKey, tier Tier, category Category, at time.Time) bool {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if entry, ok := b.pending[key]; ok {
|
||||
entry.count++
|
||||
entry.lastSeen = at
|
||||
b.pending[key] = entry
|
||||
return true
|
||||
}
|
||||
if len(b.pending) >= b.max {
|
||||
return false
|
||||
}
|
||||
b.pending[key] = counterEntry{tier: tier, category: category, count: 1, firstSeen: at, lastSeen: at}
|
||||
return true
|
||||
}
|
||||
|
||||
// len 返回待落盘的键数（≤ max）。
|
||||
func (b *counterBuffer) len() int {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return len(b.pending)
|
||||
}
|
||||
|
||||
// drain 取走全部待落盘项并清空缓冲（锁内只搬运，不 IO）。
|
||||
func (b *counterBuffer) drain() map[counterKey]counterEntry {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if len(b.pending) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := b.pending
|
||||
b.pending = make(map[counterKey]counterEntry, b.max)
|
||||
return out
|
||||
}
|
||||
|
||||
// restore 把一次**落盘失败**的项回填进缓冲，使计数不在瞬时 DB 故障里丢掉。
|
||||
// 回填同样受 `max` 约束：装不下的项被丢弃并返回丢弃数（调用方记 WARN）——
|
||||
// 「尽力而为」与「有界」冲突时，**有界**优先（§4.3 的背压是硬约束）。
|
||||
func (b *counterBuffer) restore(entries map[counterKey]counterEntry) int {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
dropped := 0
|
||||
for key, entry := range entries {
|
||||
if existing, ok := b.pending[key]; ok {
|
||||
// 落盘失败期间同一键可能又累加了新计数：合并而不是覆盖。
|
||||
existing.count += entry.count
|
||||
if entry.lastSeen.After(existing.lastSeen) {
|
||||
existing.lastSeen = entry.lastSeen
|
||||
}
|
||||
if entry.firstSeen.Before(existing.firstSeen) {
|
||||
existing.firstSeen = entry.firstSeen
|
||||
}
|
||||
b.pending[key] = existing
|
||||
continue
|
||||
}
|
||||
if len(b.pending) >= b.max {
|
||||
dropped++
|
||||
continue
|
||||
}
|
||||
b.pending[key] = entry
|
||||
}
|
||||
return dropped
|
||||
}
|
||||
|
||||
// recordCounter 把一帧计入内存聚合（design/26 §4.3：T1 与**全部帧**都计数）。
|
||||
//
|
||||
// 它**不返回错误**：计数是尽力而为，绝不让入站流水线因计数而失败。缓冲满时先尝试
|
||||
// 整批落盘；落盘失败记 WARN（DB 出问题时每次 Record 都会试一次，这也让「DB 恢复后
|
||||
// 自动收敛」不需要额外的重试器）。被吞掉的新键记 DEBUG —— 单个键的丢弃是上述 WARN
|
||||
// 的后果，不在这里重复刷 WARN（避免 DB 长时间不可用时日志洪泛）。
|
||||
func (s *SQLStore) recordCounter(ctx context.Context, ev Event) {
|
||||
if s.counters.len() >= s.counters.max {
|
||||
if err := s.flushCounters(ctx); err != nil {
|
||||
s.log.Warn("channel event counter flush failed",
|
||||
"err", err, "pending", s.counters.len(), "max_pending", s.counters.max)
|
||||
}
|
||||
}
|
||||
key := counterKey{platform: ev.Platform, accountID: ev.AccountID, platformType: ev.PlatformType}
|
||||
if !s.counters.add(key, ev.Tier, ev.Category, ev.CreatedAt) {
|
||||
s.log.Debug("channel event counter key dropped: buffer full",
|
||||
"platform", ev.Platform, "platform_type", ev.PlatformType)
|
||||
}
|
||||
}
|
||||
|
||||
// Flush 把内存里待落盘的计数写进库（组合根的常驻 ticker 与退出前的最后一次调用
|
||||
// 都走它）。没有待落盘项时是空操作。
|
||||
func (s *SQLStore) Flush(ctx context.Context) error {
|
||||
return s.flushCounters(ctx)
|
||||
}
|
||||
|
||||
// flushCounters 是 Flush 的实现：取走 → 落盘 → 失败回填。
|
||||
func (s *SQLStore) flushCounters(ctx context.Context) error {
|
||||
pending := s.counters.drain()
|
||||
if len(pending) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := s.writeCounters(ctx, pending); err != nil {
|
||||
if dropped := s.counters.restore(pending); dropped > 0 {
|
||||
s.log.Warn("channel event counter entries dropped after failed flush",
|
||||
"dropped", dropped, "max_pending", s.counters.max)
|
||||
}
|
||||
return fmt.Errorf("events: flush 计数: %w", err)
|
||||
}
|
||||
s.log.Debug("channel event counters flushed", "keys", len(pending))
|
||||
return nil
|
||||
}
|
||||
|
||||
// upsertCounterSQL 是批量落盘的唯一语句：单事务 + 同一条语句反复执行，避免 N 次
|
||||
// 往返以外的组织开销（design/26 §4.3）。`count = count + excluded.count` 使同键
|
||||
// 累加；tier/category 只在**插入**时写入，冲突时保持首次的归类（首次权威）。
|
||||
const upsertCounterSQL = `INSERT INTO channel_event_counters
|
||||
(platform, account_id, tier, platform_type, category, count, first_seen_at, last_seen_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(platform, account_id, platform_type) DO UPDATE SET
|
||||
count = count + excluded.count,
|
||||
last_seen_at = excluded.last_seen_at`
|
||||
|
||||
// writeCounters 在一个事务里 upsert 整批计数：要么全落，要么全不落（失败时由
|
||||
// flushCounters 回填缓冲，交给下一次 flush 重试）。
|
||||
func (s *SQLStore) writeCounters(ctx context.Context, pending map[counterKey]counterEntry) error {
|
||||
tx, err := s.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("events: 计数落盘开事务: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
for key, entry := range pending {
|
||||
if _, err := tx.ExecContext(ctx, upsertCounterSQL,
|
||||
key.platform, key.accountID, string(entry.tier), key.platformType, string(entry.category),
|
||||
entry.count, sqliteTimeStamp(entry.firstSeen), sqliteTimeStamp(entry.lastSeen)); err != nil {
|
||||
return fmt.Errorf("events: 计数落盘(platform=%s type=%s): %w", key.platform, key.platformType, err)
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("events: 计数落盘提交: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// listCountersSQL 的列显式列出（禁 `SELECT *`），排序固定，便于客户端做稳定展示。
|
||||
const listCountersSQL = `SELECT platform, account_id, tier, platform_type, category, count,
|
||||
first_seen_at, last_seen_at
|
||||
FROM channel_event_counters`
|
||||
|
||||
// Counters 返回逐类计数汇总，按 `(platform, account_id, platform_type)` 升序。
|
||||
// platform 为空表示不过滤（返回全部平台）。
|
||||
//
|
||||
// 它读的是**已落盘**的表，不含内存里尚未 flush 的部分 —— 这正是「计数为近似值」
|
||||
// 的来源，查询面必须如实标注（design/26 §4.3/§4.5）。需要新鲜数据时先 Flush。
|
||||
func (s *SQLStore) Counters(ctx context.Context, platform string) ([]Counter, error) {
|
||||
query := listCountersSQL
|
||||
args := make([]any, 0, 1)
|
||||
if platform != "" {
|
||||
query += " WHERE platform = ?"
|
||||
args = append(args, platform)
|
||||
}
|
||||
query += " ORDER BY platform, account_id, platform_type"
|
||||
|
||||
rows, err := s.db.QueryContext(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("events: 查计数行: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
out := make([]Counter, 0, 16)
|
||||
for rows.Next() {
|
||||
var (
|
||||
c Counter
|
||||
tier, category string
|
||||
first, last string
|
||||
)
|
||||
if err := rows.Scan(&c.Platform, &c.AccountID, &tier, &c.PlatformType, &category,
|
||||
&c.Count, &first, &last); err != nil {
|
||||
return nil, fmt.Errorf("events: 读计数行: %w", err)
|
||||
}
|
||||
c.Tier = Tier(tier)
|
||||
c.Category = Category(category)
|
||||
at, ok := parseSQLiteTime(first)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("events: 计数行 %s/%s 的 first_seen_at 不可解析(%q): %w",
|
||||
c.Platform, c.PlatformType, first, ErrBadTimestamp)
|
||||
}
|
||||
c.FirstSeenAt = at
|
||||
if ts, ok := parseSQLiteTime(last); ok {
|
||||
c.LastSeenAt = ts
|
||||
} else {
|
||||
c.LastSeenAt = at
|
||||
}
|
||||
out = append(out, c)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("events: 遍历计数行: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
Reference in new issue
Block a user