feat(backthink): 自排的成本可见面(C2 读侧,Y3)
- GET /api/v1/admin/backthink 加可选 ?trigger=(闭集=内建触发标签 + 本服务订阅的事件主题;
未知/空 ⇒ 400 invalid_trigger,不静默忽略;合法值下推到 SQL ⇒ limit 是匹配条数)
- 新端点 GET /api/v1/admin/backthink/self-wake(writeAuth):闸的只读镜像
{available,running,reason?,configFile?,gate{pending,used,limit,windowSec,default*,*IsDefault}}
三态分开(未接线 / 有配置未运行 / 运行中),读它**不消耗额度**
- 零新 env 键;旋钮源 = $DATA_DIR/config/backthink.json(**写面未做**,前置=组合根接线,已登记)
- 键名沿用设计与现有代码的 selfWakePerWindow(派工书写错成 selfWakeLimit,未改键名以免废掉主人已写的配置)
- 5 个变异(未知值静默忽略/关掉下推/恒报运行中/白名单摘键/前端把看不到画成 0/0)
⚠️ 本片改了 4 个文件面外文件(admin_query.go/routes.go/api.ts/adminSystem.ts)——已逐条说明理由
This commit is contained in:
1 parent
215eedf8e9
commit
4213b268fe
13 files changed
+1289
-43
No files matched your search
@@ -44,7 +44,6 @@ import (
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
)
|
||||
|
||||
@@ -285,11 +284,11 @@ func (b *Backthink) List(ctx context.Context, limit int) ([]Record, error) {
|
||||
return b.store.List(ctx, limit)
|
||||
}
|
||||
|
||||
// triggers 返回本服务为事件驱动思考所监听的总线主题。
|
||||
// 它是一个方法(而非包级变量),因此不会出现包级全局
|
||||
// 切片,符合编码约束。
|
||||
// triggers 返回本服务为事件驱动思考所监听的总线主题。它是
|
||||
// `EventSubjects`(trigger_domain.go 的唯一真源)的转述 —— 订阅表与 admin
|
||||
// 的 `?trigger=` 筛选白名单共用同一份,新增主题时两边同时生效。
|
||||
func (b *Backthink) triggers() []string {
|
||||
return []string{iot.DeviceUpdateSubject, SubjectReminderTrigger, SubjectSystemEvent}
|
||||
return EventSubjects()
|
||||
}
|
||||
|
||||
// subscribe 为每个触发主题注册一个处理器。每个订阅都
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
package backthink
|
||||
|
||||
import "time"
|
||||
|
||||
// 本文件是自排闸的**只读快照**出口(design/12 §N C.5/C.2 的读侧,切片 C2)。
|
||||
//
|
||||
// 为什么需要它:面板此前**看不到**「本窗口还能自排几次」—— 那个数字只出现在
|
||||
// 被拒绝那一轮的审计文案里(`backthink.self_wake.capped` 的 result),也就是
|
||||
// 「已经被拒一次」之后才第一次可见。C.5 的闸本来就维护了一份**只为上报**的镜像
|
||||
// (selfWakeGate.granted),本文件把它接出来,判定仍然只由 limiter 做
|
||||
// (§C.5:「判定始终由 limiter 做;两者用同一个时间源与同一个窗口」)。
|
||||
//
|
||||
// 🔴 本文件**不做**任何写入、不碰滑窗件(`session.RateLimiter` 至今没有用量读口,
|
||||
// 属别人的文件面):镜像在 selfWakeGate 里已经有了,读它即可。
|
||||
|
||||
// SelfWakeStatus 是自排闸的只读快照:本窗口已用多少、上限多少、窗口多长,
|
||||
// 以及容量 1 的待跑槽是否被占用。
|
||||
//
|
||||
// 它**不参与任何判定**(判定始终由滑窗与待跑槽做),因此读它永远不改变闸的状态。
|
||||
type SelfWakeStatus struct {
|
||||
// Running 报告自排是否真的能排:未 Start / 已 Stop ⇒ false(此刻 WakeNow
|
||||
// 会返回 ErrSelfWakeUnavailable)。面板据此显示「未运行」而不是「0 次」。
|
||||
Running bool
|
||||
// Pending 报告容量 1 的待跑槽是否已占用(有一轮自排排着队还没跑)。
|
||||
Pending bool
|
||||
// Used 是本窗口已用额度,与收据 `SelfWakeReceipt.Used` **同一口径**
|
||||
// (同一次取数:过期的记录先被丢弃,再计数)。
|
||||
Used int
|
||||
// Limit 与 WindowSec 是**进程生效**的上限与窗口(已 normalize;`≤0` 的那份
|
||||
// 配置不会走到这里,见 cfg.go 的 normalize 注释)。
|
||||
Limit int
|
||||
WindowSec int
|
||||
// DefaultLimit / DefaultWindowSec 是代码默认值;LimitIsDefault / WindowIsDefault
|
||||
// 报告生效值**恰好等于**默认值。
|
||||
//
|
||||
// ⚠️ 它**不是**「配置文件里没写这个键」的断言(与 config-facts 的 isDefault
|
||||
// 同一条诚实口径:网关不读配置文件,也无从区分「配了 4」与「没配」)。
|
||||
DefaultLimit int
|
||||
DefaultWindowSec int
|
||||
LimitIsDefault bool
|
||||
WindowIsDefault bool
|
||||
}
|
||||
|
||||
// SelfWakeStatus 取一张自排闸快照。第二个返回值为 false 表示**闸不在**
|
||||
// (只有零值 / 未经 `New` 构造的 Backthink 会这样)—— 调用方**不得**把零值
|
||||
// 当成事实(不伪造「0 / 0」)。
|
||||
func (b *Backthink) SelfWakeStatus() (SelfWakeStatus, bool) {
|
||||
gate, running := b.wakeGate()
|
||||
if gate == nil {
|
||||
return SelfWakeStatus{}, false
|
||||
}
|
||||
gate.mu.Lock()
|
||||
// usedLocked 顺手丢弃过期记录,因此这里读到的与判定用的是同一个口径。
|
||||
used := gate.usedLocked()
|
||||
pending := gate.pending != nil
|
||||
limit, window := gate.limit, gate.window
|
||||
gate.mu.Unlock()
|
||||
|
||||
return SelfWakeStatus{
|
||||
Running: running,
|
||||
Pending: pending,
|
||||
Used: used,
|
||||
Limit: limit,
|
||||
WindowSec: int(window / time.Second),
|
||||
DefaultLimit: defaultSelfWakeLimit,
|
||||
DefaultWindowSec: int(defaultSelfWakeWindow / time.Second),
|
||||
LimitIsDefault: limit == defaultSelfWakeLimit,
|
||||
WindowIsDefault: window == defaultSelfWakeWindow,
|
||||
}, true
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
package backthink
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 本文件守切片 C2 的**用量可见面**(design/12 §N C.5 的读侧):自排闸的只读快照
|
||||
// 必须与工具收据(SelfWakeReceipt)**同一口径**(同一个时间源、同一个窗口、
|
||||
// 同一个「消耗之后取值」),并且**读它不改变闸的状态**。
|
||||
|
||||
// TestSelfWakeStatus_MatchesReceiptAndTracksPendingSlot 是主判据:
|
||||
// 排一轮后 `Used` 必须等于收据里的 Used,待跑槽随之变占用;额度用尽后
|
||||
// `Used` 停在 `Limit`(不会因为一次拒绝而虚增)。
|
||||
//
|
||||
// 变异:把 SelfWakeStatus 的 Used 硬编码成 0(或改成 `len(granted)` 而不丢弃
|
||||
// 过期记录)⇒ 本用例在 `Used != receipt.Used` 处变红。
|
||||
func TestSelfWakeStatus_MatchesReceiptAndTracksPendingSlot(t *testing.T) {
|
||||
t.Parallel()
|
||||
// worker 卡住 ⇒ 排进去的那一轮不会把槽取走,`pending` 因此可确定地断言。
|
||||
w := newWakeThinker()
|
||||
w.holdSelfWake = make(chan struct{})
|
||||
b, _ := startSelfWake(t, Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, w)
|
||||
|
||||
// 启动即有一轮启动轮在跑(tickLoop 上的 TickImmediately),但它不占自排槽。
|
||||
before := mustStatus(t, b)
|
||||
if !before.Running {
|
||||
t.Fatal("已 Start 的服务其快照必须报告 Running=true")
|
||||
}
|
||||
if before.Used != 0 || before.Pending {
|
||||
t.Fatalf("刚启动的快照 = %+v, want Used=0 / Pending=false", before)
|
||||
}
|
||||
if before.Limit != 2 || before.WindowSec != 3600 {
|
||||
t.Fatalf("快照口径 = %+v, want Limit=2 / WindowSec=3600", before)
|
||||
}
|
||||
if before.LimitIsDefault {
|
||||
t.Fatalf("上限 2 不该被算成默认值:%+v", before)
|
||||
}
|
||||
// 窗口仍取了代码默认(1 小时)⇒ WindowIsDefault 就是 true(这是「生效值恰好等于
|
||||
// 默认值」如实上报,而不是「配置文件没写」的断言)。
|
||||
if !before.WindowIsDefault {
|
||||
t.Fatalf("窗口取了默认值却没标记:%+v", before)
|
||||
}
|
||||
if before.DefaultLimit != defaultSelfWakeLimit || before.DefaultWindowSec != int(defaultSelfWakeWindow/time.Second) {
|
||||
t.Fatalf("默认值口径 = %+v, want %d / %d", before,
|
||||
defaultSelfWakeLimit, int(defaultSelfWakeWindow/time.Second))
|
||||
}
|
||||
|
||||
receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "排一轮"})
|
||||
if err != nil {
|
||||
t.Fatalf("WakeNow: %v", err)
|
||||
}
|
||||
if !receipt.Accepted {
|
||||
t.Fatalf("收据 = %+v, want Accepted", receipt)
|
||||
}
|
||||
after := mustStatus(t, b)
|
||||
if after.Used != receipt.Used || after.Used != 1 {
|
||||
t.Fatalf("快照 Used = %d, 收据 Used = %d, want 1(同一口径)", after.Used, receipt.Used)
|
||||
}
|
||||
if !after.Pending {
|
||||
t.Fatalf("排入后待跑槽应报告占用:%+v", after)
|
||||
}
|
||||
|
||||
// 第二次撞上满槽 ⇒ 合并(不消耗额度),快照的 Used 不得变。
|
||||
if _, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "再排一轮"}); err != nil {
|
||||
t.Fatalf("WakeNow 合并: %v", err)
|
||||
}
|
||||
merged := mustStatus(t, b)
|
||||
if merged.Used != 1 {
|
||||
t.Fatalf("合并后 Used = %d, want 1(合并不额外占额度)", merged.Used)
|
||||
}
|
||||
if merged.Limit != 2 {
|
||||
t.Fatalf("合并后 Limit = %d, want 2", merged.Limit)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSelfWakeStatus_StoppedServiceIsNotRunning 钉住「未运行」这一态:
|
||||
// 服务在跑之前(以及 Stop 之后)快照必须如实报 Running=false,好让面板显示
|
||||
// 「此刻排不出新的一轮」而不是「已用 0 次」。
|
||||
//
|
||||
// 变异:把 SelfWakeStatus 的 Running 直接写 true ⇒ 本用例变红。
|
||||
func TestSelfWakeStatus_StoppedServiceIsNotRunning(t *testing.T) {
|
||||
t.Parallel()
|
||||
b, _ := newBackthink(t, Config{}, Deps{Thinker: newWakeThinker(), State: &mockState{}})
|
||||
|
||||
st := mustStatus(t, b)
|
||||
if st.Running {
|
||||
t.Fatalf("未 Start 的快照 = %+v, want Running=false", st)
|
||||
}
|
||||
// 未运行时额度必然是 0,且上限已 normalize(面板仍能看到「上限 4/小时」)。
|
||||
if st.Used != 0 {
|
||||
t.Fatalf("未运行时 Used = %d, want 0", st.Used)
|
||||
}
|
||||
if st.Limit != defaultSelfWakeLimit || !st.LimitIsDefault || !st.WindowIsDefault {
|
||||
t.Fatalf("零配置的快照 = %+v, want 默认口径 %d/%v", st,
|
||||
defaultSelfWakeLimit, defaultSelfWakeWindow)
|
||||
}
|
||||
|
||||
b.Start(context.Background())
|
||||
if !mustStatus(t, b).Running {
|
||||
t.Fatal("Start 之后快照必须报告 Running=true")
|
||||
}
|
||||
if err := b.Stop(context.Background()); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
if mustStatus(t, b).Running {
|
||||
t.Fatal("Stop 之后快照必须报告 Running=false")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSelfWakeStatus_ReadDoesNotConsumeBudget 钉住「读是纯读」:反复读快照
|
||||
// 不能让额度减少,也不能让待跑槽变化。
|
||||
//
|
||||
// 变异:把快照实现改成调用 limiter.Allow()(成功即消耗)⇒ 本用例在 Used 处变红。
|
||||
func TestSelfWakeStatus_ReadDoesNotConsumeBudget(t *testing.T) {
|
||||
t.Parallel()
|
||||
b, _ := newBackthink(t, Config{SelfWakeLimit: 1, SelfWakeWindow: time.Hour},
|
||||
Deps{Thinker: newWakeThinker(), State: &mockState{}})
|
||||
b.Start(context.Background())
|
||||
t.Cleanup(func() { _ = b.Stop(context.Background()) })
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
if got := mustStatus(t, b).Used; got != 0 {
|
||||
t.Fatalf("第 %d 次读快照后 Used = %d, want 0(读不改闸)", i+1, got)
|
||||
}
|
||||
}
|
||||
// 读完之后额度仍然可用:唯一那一格必须能真排上一轮。
|
||||
receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "读完之后再排"})
|
||||
if err != nil {
|
||||
t.Fatalf("WakeNow: %v", err)
|
||||
}
|
||||
if !receipt.Accepted || receipt.Used != 1 {
|
||||
t.Fatalf("收据 = %+v, want Accepted 且 Used=1(读快照没有偷走额度)", receipt)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSelfWakeStatus_UnreadableOnlyWhenGateMissing 钉住第二个返回值的诚实语义:
|
||||
// 正常构造的服务一定有闸(New 里建),只有零值 Backthink 才报告「看不到」。
|
||||
//
|
||||
// 变异:把 `return SelfWakeStatus{}, false` 改成 `true` ⇒ 本用例变红
|
||||
// (调用方会把 0/0 当成真实用量)。
|
||||
func TestSelfWakeStatus_UnreadableOnlyWhenGateMissing(t *testing.T) {
|
||||
t.Parallel()
|
||||
if _, ok := (&Backthink{}).SelfWakeStatus(); ok {
|
||||
t.Fatal("零值 Backthink 没有闸,必须报告 ok=false(不得伪造 0/0)")
|
||||
}
|
||||
b, _ := newBackthink(t, Config{}, Deps{Thinker: newWakeThinker(), State: &mockState{}})
|
||||
if _, ok := b.SelfWakeStatus(); !ok {
|
||||
t.Fatal("经 New 构造的服务一定有闸,必须报告 ok=true")
|
||||
}
|
||||
}
|
||||
|
||||
// mustStatus 取快照并在「闸不在」时直接失败(正常构造的服务不该走到那一步)。
|
||||
func mustStatus(t *testing.T, b *Backthink) SelfWakeStatus {
|
||||
t.Helper()
|
||||
st, ok := b.SelfWakeStatus()
|
||||
if !ok {
|
||||
t.Fatal("SelfWakeStatus 报告闸不在:正常构造的服务不该这样")
|
||||
}
|
||||
return st
|
||||
}
|
||||
@@ -45,6 +45,14 @@ ALTER TABLE backthink_records ADD COLUMN urgency TEXT NOT NULL DEFAULT ''`
|
||||
const createRecordsIndexSQL = `
|
||||
CREATE INDEX IF NOT EXISTS idx_backthink_records_at ON backthink_records(at)`
|
||||
|
||||
// createRecordsTriggerIndexSQL 支撑按触发源筛选的那条查询
|
||||
// (`WHERE trigger = ? ORDER BY id DESC`)。索引条目是 (trigger, rowid),而主键
|
||||
// `id` 就是 rowid ⇒ SQLite 能直接用一次索引范围扫描(反向)同时满足筛选与排序。
|
||||
// 它是**本片新增读路径**的配套:没有它,筛选就是全表扫描 + 排序(表按心跳每 15 分钟
|
||||
// 增长一行,一年就是三万余行)。
|
||||
const createRecordsTriggerIndexSQL = `
|
||||
CREATE INDEX IF NOT EXISTS idx_backthink_records_trigger ON backthink_records(trigger)`
|
||||
|
||||
const insertRecordSQL = `
|
||||
INSERT INTO backthink_records
|
||||
(trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency)
|
||||
@@ -56,6 +64,19 @@ FROM backthink_records
|
||||
ORDER BY id DESC
|
||||
LIMIT ?`
|
||||
|
||||
// selectRecordsByTriggerSQL 与 selectRecordsSQL 逐字同形,只多一个按
|
||||
// `Record.Trigger` 的**服务端**筛选(参数化,不拼字符串)。
|
||||
//
|
||||
// 为什么下推到 SQL 而不是「取一页再在内存里筛」:那样 `limit` 就变成「这一页里
|
||||
// 恰好匹配的条数」,面板会看到「自我唤醒 0 条」而库里其实有 —— 一个静默的假空值
|
||||
// (design/12 §N C.6 的服务端筛选要的正是「筛不到 ≠ 没发生」)。
|
||||
const selectRecordsByTriggerSQL = `
|
||||
SELECT id, trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency
|
||||
FROM backthink_records
|
||||
WHERE trigger = ?
|
||||
ORDER BY id DESC
|
||||
LIMIT ?`
|
||||
|
||||
// recordsTableName 是本模块自建表的物理名(DDL 与判列共用一处)。
|
||||
const recordsTableName = "backthink_records"
|
||||
|
||||
@@ -115,6 +136,9 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error)
|
||||
if _, err := db.ExecContext(ctx, createRecordsIndexSQL); err != nil {
|
||||
return nil, fmt.Errorf("backthink: ensure backthink_records index: %w", err)
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, createRecordsTriggerIndexSQL); err != nil {
|
||||
return nil, fmt.Errorf("backthink: ensure backthink_records trigger index: %w", err)
|
||||
}
|
||||
return &Store{db: db, log: log}, nil
|
||||
}
|
||||
|
||||
@@ -207,9 +231,7 @@ func (s *Store) Append(ctx context.Context, r Record) (int64, error) {
|
||||
// defaultListLimit。urgency 列的空串如实读成「未知」(见
|
||||
// Record.Urgency),不做任何补齐或推断。
|
||||
func (s *Store) List(ctx context.Context, limit int) ([]Record, error) {
|
||||
if limit <= 0 {
|
||||
limit = defaultListLimit
|
||||
}
|
||||
limit = normListLimit(limit)
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -218,8 +240,40 @@ func (s *Store) List(ctx context.Context, limit int) ([]Record, error) {
|
||||
return nil, fmt.Errorf("backthink: list records: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
return scanRecords(rows, limit)
|
||||
}
|
||||
|
||||
records := make([]Record, 0, limit)
|
||||
// ListByTrigger 返回**某一触发源**的最近记录,最新的在前:语义与 List 逐字相同,
|
||||
// 只多一个参数化的 `trigger = ?`(design/12 §N C.6 的服务端筛选)。
|
||||
//
|
||||
// ⚠️ 它按**原始落库值**精确匹配(不模糊、不前缀):`Record.Trigger` 对事件轮是事件
|
||||
// 主题字符串,对内置轮是 `heartbeat` / `startup` / `self_wake` —— 合法取值由
|
||||
// `IsTriggerValue`(trigger_domain.go 的唯一真源)判定,本层不做二次裁剪。
|
||||
func (s *Store) ListByTrigger(ctx context.Context, trigger string, limit int) ([]Record, error) {
|
||||
limit = normListLimit(limit)
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rows, err := s.db.QueryContext(ctx, selectRecordsByTriggerSQL, trigger, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("backthink: list records by trigger: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
return scanRecords(rows, limit)
|
||||
}
|
||||
|
||||
// normListLimit 把非正的 limit 归一为 defaultListLimit(两条读路径同一口径)。
|
||||
func normListLimit(limit int) int {
|
||||
if limit <= 0 {
|
||||
return defaultListLimit
|
||||
}
|
||||
return limit
|
||||
}
|
||||
|
||||
// scanRecords 是两条读路径共用的行解码(抽出来是为了让「列序 / 空串语义」只有
|
||||
// 一处真源:再加第三条查询时不会悄悄错位)。
|
||||
func scanRecords(rows *sql.Rows, capacity int) ([]Record, error) {
|
||||
records := make([]Record, 0, capacity)
|
||||
for rows.Next() {
|
||||
var (
|
||||
r Record
|
||||
|
||||
@@ -127,7 +127,7 @@ type Trigger struct {
|
||||
// heartbeatTrigger 是周期性、低重要度的触发。低重要度意味着
|
||||
// 该轮很少打扰用户,遵守 "默认不发送 chat 类型消息"。
|
||||
func heartbeatTrigger() Trigger {
|
||||
return Trigger{Source: "heartbeat", Summary: "periodic autonomous reflection", Import: ImportanceLow, Kind: TriggerHeartbeat}
|
||||
return Trigger{Source: HeartbeatSource, Summary: "periodic autonomous reflection", Import: ImportanceLow, Kind: TriggerHeartbeat}
|
||||
}
|
||||
|
||||
// StartupSummary 是启动轮思考(裁决 07.4.4,design/12 §1)的触发摘要。它同时是
|
||||
@@ -139,7 +139,7 @@ const StartupSummary = "process startup"
|
||||
// Kind 仍为 TriggerHeartbeat:启动轮的推送与心跳走完全相同的门与限频
|
||||
// (PushGate / 打扰级联 / 活动门),进程反复重启也不会刷屏主人。
|
||||
func startupTrigger() Trigger {
|
||||
return Trigger{Source: "startup", Summary: StartupSummary, Import: ImportanceLow, Kind: TriggerHeartbeat}
|
||||
return Trigger{Source: StartupSource, Summary: StartupSummary, Import: ImportanceLow, Kind: TriggerHeartbeat}
|
||||
}
|
||||
|
||||
// triggerForEvent 把总线事件映射为 Trigger:主题给出可读摘要与主题级默认重要度,
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
package backthink
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
|
||||
)
|
||||
|
||||
// 本文件是「触发源取值域」的**唯一真源**:`Record.Trigger`(= `Trigger.Source` 字符串)
|
||||
// 可能出现的取值,以及本服务订阅哪些事件主题。
|
||||
//
|
||||
// 为什么要独立出来:admin 的 `GET /api/v1/admin/backthink?trigger=` **服务端筛选**
|
||||
// (design/12 §N C.6,切片 C2)必须**如实拒绝未知取值**(400),而「未知」只能相对
|
||||
// 一个闭集判断。若在网关里另抄一份主题清单,新增订阅主题时就会出现「数据里有这个取值、
|
||||
// 筛选却报错」的静默漂移 —— 因此订阅表与筛选白名单从同一处取。
|
||||
//
|
||||
// ⚠️ 与 design/12 §C.6「`trigger` 的取值域不封闭」**不矛盾**:那条约束的是**显示**口径
|
||||
// (客户端按字面值显示、不做闭集校验、不隐藏不认识的取值);这里定义的是**筛选白名单**
|
||||
// —— 面向的是「筛选参数打错了要报错,而不是静默返回一个空列表」(静默的空列表会被
|
||||
// 读成「这段时间没有这一类的思考」,那正是本片要避免的假事实)。
|
||||
|
||||
// 内建触发标签。它们是 `Trigger.Source` 的字面值,也是 `Record.Trigger` 的落库值,
|
||||
// 生产构造点分别是 heartbeatTrigger / startupTrigger(trigger.go)。
|
||||
const (
|
||||
// HeartbeatSource 是周期性自主反思(含启动轮之外的每次心跳)的标签。
|
||||
HeartbeatSource = "heartbeat"
|
||||
// StartupSource 是进程启动后第一轮思考的标签(裁决 07.4.4)。
|
||||
StartupSource = "startup"
|
||||
)
|
||||
|
||||
// EventSubjects 返回本服务订阅的事件主题 —— **唯一真源**:`subscribe` 用它注册订阅,
|
||||
// `TriggerFilterValues` 用它拼筛选白名单,因此新增一个订阅主题时两处同时生效
|
||||
// (不会出现「数据库里有新主题的记录、筛选却 400」的漂移)。
|
||||
func EventSubjects() []string {
|
||||
return []string{iot.DeviceUpdateSubject, SubjectReminderTrigger, SubjectSystemEvent}
|
||||
}
|
||||
|
||||
// TriggerFilterValues 返回 `Record.Trigger` 可能出现的**全部取值**(闭集):
|
||||
// 三个内建标签(心跳 / 启动 / 自排)+ 订阅的事件主题。
|
||||
//
|
||||
// 它是 admin 筛选参数的白名单,也是「面板可选项」的真源;顺序稳定,便于报错文案
|
||||
// 与客户端下拉复用。
|
||||
func TriggerFilterValues() []string {
|
||||
return append([]string{HeartbeatSource, StartupSource, SelfWakeSource}, EventSubjects()...)
|
||||
}
|
||||
|
||||
// IsTriggerValue 报告 v 是否是 `Record.Trigger` 可能出现的取值(白名单判定)。
|
||||
func IsTriggerValue(v string) bool {
|
||||
for _, x := range TriggerFilterValues() {
|
||||
if x == v {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ListByTrigger 按触发源读取记录(服务端筛选的读口,design/12 §N C.6 / 切片 C2)。
|
||||
// 它把筛选**下推**到存储层:`limit` 因此是「**匹配**的记录条数」,而不是
|
||||
// 「先取一页再筛掉不匹配的」(后者会给面板一个静默的假空值)。
|
||||
//
|
||||
// 委托给 `*Store`(自己不认识 SQL);`List` 的语义、排序、空串口径一字不改。
|
||||
func (b *Backthink) ListByTrigger(ctx context.Context, trigger string, limit int) ([]Record, error) {
|
||||
return b.store.ListByTrigger(ctx, trigger, limit)
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
package backthink
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 本文件守切片 C2 的两条读侧能力(design/12 §N C.5/C.6):
|
||||
// 1. **触发源取值域**(trigger_domain.go 的唯一真源)—— 内建标签 + 订阅的事件主题,
|
||||
// 它是 admin `?trigger=` 筛选白名单的依据;
|
||||
// 2. **按触发源读取**(ListByTrigger 下推到 SQL)—— `limit` 必须是**匹配**条数,
|
||||
// 而不是「先取一页再筛掉不匹配的」(后者会给面板一个静默的假空值)。
|
||||
|
||||
// TestTriggerFilterValues_CoversBuiltinsAndSubscribedSubjects 钉住白名单的**构成**:
|
||||
// 三个内建标签(心跳 / 启动 / 自排)+ 本服务订阅的全部事件主题。
|
||||
//
|
||||
// 变异:把 SelfWakeSource 从列表里摘掉 ⇒ 本用例变红(面板将筛不到自排轮,
|
||||
// 而它恰恰是这一档最想看的);少列一个订阅主题 ⇒ 同红。
|
||||
func TestTriggerFilterValues_CoversBuiltinsAndSubscribedSubjects(t *testing.T) {
|
||||
t.Parallel()
|
||||
values := TriggerFilterValues()
|
||||
set := map[string]bool{}
|
||||
for _, v := range values {
|
||||
if set[v] {
|
||||
t.Fatalf("白名单里有重复取值 %q:%v", v, values)
|
||||
}
|
||||
set[v] = true
|
||||
}
|
||||
for _, want := range []string{HeartbeatSource, StartupSource, SelfWakeSource} {
|
||||
if !set[want] {
|
||||
t.Fatalf("白名单缺内建标签 %q:%v", want, values)
|
||||
}
|
||||
}
|
||||
subjects := EventSubjects()
|
||||
if len(subjects) == 0 {
|
||||
t.Fatal("订阅主题表为空:白名单会漏掉全部事件轮")
|
||||
}
|
||||
for _, s := range subjects {
|
||||
if !set[s] {
|
||||
t.Fatalf("白名单缺订阅主题 %q:%v", s, values)
|
||||
}
|
||||
}
|
||||
// 反向:白名单里的每个取值都必须真的可能出现在 Record.Trigger 上
|
||||
// (= 内建标签 或 订阅主题)—— 多列一个「不可能出现」的值会让筛选永远返回空。
|
||||
for _, v := range values {
|
||||
if v == HeartbeatSource || v == StartupSource || v == SelfWakeSource {
|
||||
continue
|
||||
}
|
||||
if !isEventSubject(v) {
|
||||
t.Fatalf("白名单含一个不可能是触发源的取值 %q:%v", v, values)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// isEventSubject 只在本测试里做反向核对(生产侧没有这个判定:白名单由
|
||||
// TriggerFilterValues 拼出来,不需要反查)。
|
||||
func isEventSubject(v string) bool {
|
||||
for _, s := range EventSubjects() {
|
||||
if s == v {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// TestIsTriggerValue_RejectsUnknown 钉住白名单判定:已知取值通行,未知/空串一律拒。
|
||||
//
|
||||
// 变异:把 IsTriggerValue 改成恒 true ⇒ 服务端筛选会「接受一切但筛不中」,
|
||||
// 网关那条 400 的用例随之变红。
|
||||
func TestIsTriggerValue_RejectsUnknown(t *testing.T) {
|
||||
t.Parallel()
|
||||
for _, ok := range []string{HeartbeatSource, StartupSource, SelfWakeSource, "iot.device_update"} {
|
||||
if !IsTriggerValue(ok) {
|
||||
t.Fatalf("IsTriggerValue(%q) = false, want true", ok)
|
||||
}
|
||||
}
|
||||
for _, bad := range []string{"", "event", "self-wake", "HEARTBEAT", "all", "iot"} {
|
||||
if IsTriggerValue(bad) {
|
||||
t.Fatalf("IsTriggerValue(%q) = true, want false", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestStore_ListByTrigger_IsServerSideFilter 钉住**下推**语义:筛选后 `limit` 数的是
|
||||
// 匹配条数(而不是「本页里恰好匹配的条数」)。
|
||||
//
|
||||
// 变异:把 ListByTrigger 改回「List 之后内存筛」的退化实现(拿 limit 条再筛)⇒
|
||||
// 下面的 `limit=1` 断言会拿到 0 条(最新一条是心跳,被筛掉后没有别的自排)⇒ 变红。
|
||||
func TestStore_ListByTrigger_IsServerSideFilter(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
s, err := NewStore(ctx, newTestDB(t), discardLog())
|
||||
if err != nil {
|
||||
t.Fatalf("NewStore: %v", err)
|
||||
}
|
||||
base := time.Date(2026, 10, 6, 12, 0, 0, 0, time.UTC)
|
||||
// 交替写入:心跳(最新)/ 自排 / 心跳 / 自排 / 心跳(最旧)。
|
||||
triggers := []string{HeartbeatSource, SelfWakeSource, HeartbeatSource, SelfWakeSource, HeartbeatSource}
|
||||
for i, tr := range triggers {
|
||||
if _, err := s.Append(ctx, Record{Trigger: tr, At: base.Add(time.Duration(i) * time.Minute)}); err != nil {
|
||||
t.Fatalf("Append %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
|
||||
got, err := s.ListByTrigger(ctx, SelfWakeSource, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("ListByTrigger: %v", err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("自排记录 = %d 条, want 2(%+v)", len(got), got)
|
||||
}
|
||||
for _, r := range got {
|
||||
if r.Trigger != SelfWakeSource {
|
||||
t.Fatalf("筛选结果含非自排行:%+v", r)
|
||||
}
|
||||
}
|
||||
// 最新的在前(与 List 同口径)。
|
||||
if got[0].ID != 4 || got[1].ID != 2 {
|
||||
t.Fatalf("顺序 = %d,%d, want 4,2(最新在前)", got[0].ID, got[1].ID)
|
||||
}
|
||||
|
||||
// 🔴 下推的关键判据:limit 数的是**匹配**条数。
|
||||
one, err := s.ListByTrigger(ctx, SelfWakeSource, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("ListByTrigger limit=1: %v", err)
|
||||
}
|
||||
if len(one) != 1 || one[0].ID != 4 {
|
||||
t.Fatalf("limit=1 结果 = %+v, want 只有 id=4(匹配条数,不是先取一页再筛)", one)
|
||||
}
|
||||
|
||||
// 未出现过的合法取值 ⇒ 空列表(不是错误)。
|
||||
none, err := s.ListByTrigger(ctx, "reminder.trigger", 10)
|
||||
if err != nil {
|
||||
t.Fatalf("ListByTrigger(reminder.trigger): %v", err)
|
||||
}
|
||||
if len(none) != 0 {
|
||||
t.Fatalf("无匹配时应返回空列表,得到 %+v", none)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBackthink_ListByTrigger_DelegatesToStore 钉住服务层的读口真的接到存储
|
||||
// (网关的可选接口 `backthinkTriggerLister` 靠的就是这个方法签名)。
|
||||
//
|
||||
// 变异:把 `Backthink.ListByTrigger` 改成忽略 trigger 直接 List ⇒ 本用例变红。
|
||||
func TestBackthink_ListByTrigger_DelegatesToStore(t *testing.T) {
|
||||
t.Parallel()
|
||||
b, _ := newBackthink(t, Config{}, Deps{Thinker: newWakeThinker(), State: &mockState{}})
|
||||
ctx := context.Background()
|
||||
for i := 0; i < 3; i++ {
|
||||
if _, err := b.store.Append(ctx, Record{Trigger: HeartbeatSource, At: time.UnixMilli(int64(i))}); err != nil {
|
||||
t.Fatalf("Append: %v", err)
|
||||
}
|
||||
}
|
||||
if _, err := b.store.Append(ctx, Record{Trigger: SelfWakeSource, At: time.UnixMilli(99)}); err != nil {
|
||||
t.Fatalf("Append: %v", err)
|
||||
}
|
||||
|
||||
got, err := b.ListByTrigger(ctx, SelfWakeSource, 50)
|
||||
if err != nil {
|
||||
t.Fatalf("ListByTrigger: %v", err)
|
||||
}
|
||||
if len(got) != 1 || got[0].Trigger != SelfWakeSource {
|
||||
t.Fatalf("got = %+v, want 只有那条自排记录", got)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user