feat(memory): T1 加保守全局速率限流(防每天数百到上千次调用失控)
主人 2026-10-07 裁定:按最优解加全局限流。 - 新增 t1RateGate(滑动窗口、可注入时钟,~70 行);默认 120 次 / 60 分钟(现网 700-1000 条新记忆/天 ≈ 25-42/小时, 取 120 留约 3-5 倍余量) - 过闸摆在「t1Enabled + 有灰区候选」两个守卫之后、发请求之前 ⇒ 不调模型就不占额度;放行才记账 - 超限语义与既有失败路径同源:跳过 T1、决定仍是 create、零改写, 只多一个 reason=t1_throttled 进留痕 + ReviewStats.T1Throttled 计数 - 配置键 t1PerWindow / t1WindowSec 放 $DATA_DIR/config/memory-tidy.json (零新 env 键,启动期读一次);坏配置回落默认 + WARN,不启动失败 - 每窗口只记一条 WARN(超限是预期行为,一天几百条会淹掉真故障) - IngestRuleVersion 仍为 d3.2:判定规则集与解析契约没变,变的只是 额度状态,reason 列已可分辨(升版会有 §17.6 记过的幂等键重复行副作用) 自研而非复用 session.NewRateLimiter 的理由(design/12 §N C.5 自己登记过 这条取舍):既有件 Allow() 内部写死 time.Now(),没有时钟注入缝, 而本切片必须能注入可控时钟;且复用会把 channel 层概念拖进 memory 包。
This commit is contained in:
1 parent
987cefdf28
commit
a684328b76
10 files changed
+782
-10
No files matched your search
@@ -39,7 +39,9 @@ var errT1ModelNotReady = errors.New("memory review: t1 judge: model adapter not
|
||||
// 🔴 两个开关都不在 env 里(`.env` 已冻结,design/13 §2.1),它们读
|
||||
// `$DATA_DIR/config/memory-tidy.json`,与整理自动档同一个文件、**零新 env 键**:
|
||||
// - `{"ingestGate": false}` ⇒ 一键回到「只插不审」(同步入库闸);
|
||||
// - `{"t1Judge": false}` ⇒ 一键回到纯规则(T1 灰区判定,**零模型调用**)。
|
||||
// - `{"t1Judge": false}` ⇒ 一键回到纯规则(T1 灰区判定,**零模型调用**);
|
||||
// - `{"t1PerWindow": 120, "t1WindowSec": 3600}` ⇒ T1 的**全局限速**额度
|
||||
// (design/23 §15.4);缺位 = 代码默认 120 次/60 分钟。
|
||||
func (a *app) wireMemoryReview() error {
|
||||
cfg, err := config.LoadMemoryReviewConfig(envPath)
|
||||
if err != nil {
|
||||
@@ -55,6 +57,10 @@ func (a *app) wireMemoryReview() error {
|
||||
// 一键回到纯规则(零模型调用)。与入库闸同一个文件、同一套「没配 = 保持默认」
|
||||
// 口径,**零新 env 键**。
|
||||
t1Off := a.memoryT1JudgeOff()
|
||||
// T1 的全局限速额度(design/23 §15.4):同样读那个文件,缺位/坏文件一律
|
||||
// 交零值给 normalizeReviewConfig 回落**代码默认值**(120 次/60 分钟)——
|
||||
// 坏配置绝不能让一处限速静默变成「无上限」。
|
||||
t1PerWindow, t1WindowSec := a.memoryT1RateLimit()
|
||||
reviewer, err := memory.NewPreInsertReviewer(memory.ReviewConfig{
|
||||
Enabled: cfg.Enabled,
|
||||
QueueSize: cfg.QueueSize,
|
||||
@@ -70,6 +76,12 @@ func (a *app) wireMemoryReview() error {
|
||||
// T1 的时长走代码默认值(8s,且被单条审查的 Timeout 收紧取小):
|
||||
// 它是**实现参数**不是运维旋钮 —— 要让 T1 完全不花钱,就关掉它。
|
||||
DisableT1: t1Off,
|
||||
// T1 的**全局限速**额度(design/23 §15.4):数值来自
|
||||
// `$DATA_DIR/config/memory-tidy.json` 的 `t1PerWindow` / `t1WindowSec`,
|
||||
// 缺位或坏配置时是零值 ⇒ 由 normalizeReviewConfig 回落代码默认
|
||||
// (120 次/60 分钟)。**没有「无上限」这个状态**:要零花费请关 t1Judge。
|
||||
T1PerWindow: t1PerWindow,
|
||||
T1Window: time.Duration(t1WindowSec) * time.Second,
|
||||
}, memory.ReviewDeps{
|
||||
Store: a.memStore,
|
||||
Embedder: a.embedder,
|
||||
@@ -114,6 +126,31 @@ func (a *app) memoryT1JudgeOff() bool {
|
||||
return tuning.T1JudgeOff()
|
||||
}
|
||||
|
||||
// memoryT1RateLimit 读 **T1 全局限速**的两项旋钮:窗口额度(次)与窗口长度(秒)。
|
||||
//
|
||||
// 口径与 memoryIngestGateOff / memoryT1JudgeOff 逐字相同:读不到/解不开
|
||||
// `memory-tidy.json` 时返回零值(= 不覆盖)并记 WARN —— 零值经
|
||||
// normalizeReviewConfig 回落**代码默认值**(120 次/60 分钟),因此坏配置既不会
|
||||
// 让限速静默变成「无上限」,也不会让 T1 静默失效。
|
||||
//
|
||||
// 只有**两项都写了且为正**才报一条 Info(自解释的配置回执);只写了一项时静默
|
||||
// 回落缺的那一项,不报「已配置」以免报出一个与生效值不符的印象。
|
||||
func (a *app) memoryT1RateLimit() (perWindow int, windowSec int) {
|
||||
tuning, path, err := a.readMemoryTidyTuning()
|
||||
if err != nil {
|
||||
logger.WithSource(a.log, "memory/review").Warn(
|
||||
"memory t1 rate limit tuning unreadable, falling back to code defaults",
|
||||
"path", path, "error", err)
|
||||
return 0, 0
|
||||
}
|
||||
if tuning.T1PerWindow > 0 && tuning.T1WindowSec > 0 {
|
||||
logger.WithSource(a.log, "memory/review").Info(
|
||||
"memory t1 rate limit loaded",
|
||||
"path", path, "t1_per_window", tuning.T1PerWindow, "t1_window_sec", tuning.T1WindowSec)
|
||||
}
|
||||
return tuning.T1PerWindow, tuning.T1WindowSec
|
||||
}
|
||||
|
||||
// memoryT1Judge 把「调用时才解析的模型适配器」包成 T1 的一次性调用口。
|
||||
//
|
||||
// 为什么是闭包而不是直接塞 `a.modelAdapter`:组合根的顺序是 **wireMemory(建
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -68,6 +70,11 @@ func TestMemoryReviewIsWired(t *testing.T) {
|
||||
"a.memoryT1JudgeOff()",
|
||||
"DisableT1: t1Off",
|
||||
"Judge: a.memoryT1Judge()",
|
||||
// T1 的**全局限速**额度(design/23 §15.4)同样来自那个文件,
|
||||
// 缺位/坏文件一律回落代码默认(120 次/60 分钟)。
|
||||
"a.memoryT1RateLimit()",
|
||||
"T1PerWindow: t1PerWindow",
|
||||
"T1Window: time.Duration(t1WindowSec) * time.Second",
|
||||
},
|
||||
"main.go": {
|
||||
"a.memoryReview.Start(ctx)",
|
||||
@@ -186,6 +193,52 @@ func TestMemoryT1Judge_ReadsDataDirConfig(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestMemoryT1RateLimit_ReadsDataDirConfig 钉住 **T1 全局限速**的额度确实来自
|
||||
// `$DATA_DIR/config/memory-tidy.json` 的 `t1PerWindow` / `t1WindowSec`
|
||||
// (design/23 §15.4):T1 默认开,每次有灰区候选的审查就发一次调用 ⇒ 额度必须能
|
||||
// 按现网节奏调,而 `.env` 已冻结、不许为它开 env 键。
|
||||
//
|
||||
// 口径与入库闸/T1 开关同一条:没配 = 返回零值(交给 normalizeReviewConfig 回落
|
||||
// 代码默认 120 次/60 分钟);坏文件 = 同样回落默认额 **并记 WARN** ——
|
||||
// 坏配置绝不能让限速静默变成「无上限」。
|
||||
func TestMemoryT1RateLimit_ReadsDataDirConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
empty := &app{cfg: &config.Config{DataDir: t.TempDir()}, log: tidyTestLog()}
|
||||
if perWindow, windowSec := empty.memoryT1RateLimit(); perWindow != 0 || windowSec != 0 {
|
||||
t.Fatalf("没配过时应当返回零值(= 回落代码默认):%d/%d", perWindow, windowSec)
|
||||
}
|
||||
|
||||
dataDir := t.TempDir()
|
||||
dir := filepath.Join(dataDir, "config")
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
t.Fatalf("mkdir: %v", err)
|
||||
}
|
||||
path := filepath.Join(dir, "memory-tidy.json")
|
||||
if err := os.WriteFile(path, []byte(`{"t1PerWindow": 240, "t1WindowSec": 1800}`), 0o600); err != nil {
|
||||
t.Fatalf("write: %v", err)
|
||||
}
|
||||
set := &app{cfg: &config.Config{DataDir: dataDir}, log: tidyTestLog()}
|
||||
if perWindow, windowSec := set.memoryT1RateLimit(); perWindow != 240 || windowSec != 1800 {
|
||||
t.Fatalf("额度没被读到:%d/%d, want 240/1800", perWindow, windowSec)
|
||||
}
|
||||
|
||||
// 坏配置:回落默认额(零值)+ 一条 WARN,绝不静默不限流。
|
||||
if err := os.WriteFile(path, []byte(`{"t1PerWindow": "many"}`), 0o600); err != nil {
|
||||
t.Fatalf("write broken: %v", err)
|
||||
}
|
||||
var buf bytes.Buffer
|
||||
broken := &app{cfg: &config.Config{DataDir: dataDir},
|
||||
log: slog.New(slog.NewTextHandler(&buf, nil))}
|
||||
if perWindow, windowSec := broken.memoryT1RateLimit(); perWindow != 0 || windowSec != 0 {
|
||||
t.Fatalf("坏配置下应当回落零值:%d/%d", perWindow, windowSec)
|
||||
}
|
||||
if logged := buf.String(); !strings.Contains(logged, "t1 rate limit tuning unreadable") ||
|
||||
!strings.Contains(logged, "level=WARN") {
|
||||
t.Fatalf("坏配置没有留下 WARN:%q", logged)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMemoryT1Judge_UsesTheWiredModelStack 钉住 T1 的模型出口**复用既有模型栈**
|
||||
// (不新造 LLM 客户端),且返回一个闭包而不是当场取值 —— 本服务在 wireMemory
|
||||
// 阶段装配,而 a.modelAdapter 要等更晚的 wireRuntime 才建好。
|
||||
|
||||
@@ -17,6 +17,12 @@
|
||||
// 是同一条 —— 它是一处**会花掉模型额度**的开关,必须能一行配置关掉(关掉即回落
|
||||
// 纯规则、零模型调用),所以同样落在这里,不新开文件家族、不加 env 键。
|
||||
//
|
||||
// 2026-10-07 同日追加 `t1PerWindow` / `t1WindowSec`(T1 的**全局限速**,
|
||||
// design/23 §15.4 主人裁定「加一层保守的全局速率限流」):T1 默认开,每次「有灰区
|
||||
// 候选的审查」就发一次调用,现网每天数百到上千条新记忆 ⇒ 那是一处**没有刹车**的
|
||||
// 额度出口。额度必须能按现网节奏调(且不许为它开 env 键),因此落在这里;
|
||||
// **零值不表示「无限制」**,只表示「没配、用代码默认值」。
|
||||
//
|
||||
// 边界与 backthink.go 同源(那是同一族的手续):
|
||||
//
|
||||
// 1. 本文件只管**文件形状 + 读取**:它不认识 `TidyConfig`,也不认识「零值回落
|
||||
@@ -87,6 +93,21 @@ type MemoryTidyTuning struct {
|
||||
// 指针也是同一个理由:代码默认值是「开」,用 bool 会让「这个键没写」与
|
||||
// 「显式写 false」不可区分。
|
||||
T1Judge *bool `json:"t1Judge"`
|
||||
// T1PerWindow 是 T1 的**全局**窗口额度(次数)—— 全进程共享一个窗口,超限
|
||||
// 即跳过 T1 并回落 `create`(`reason=t1_throttled`,design/23 §15.4)。
|
||||
// 零/负 = 没配 ⇒ 用代码默认值(memory 侧的 defaultT1PerWindow)。
|
||||
//
|
||||
// **刻意没有「无上限」语义**(与 autoApplyPerRound 同一口径):≤0 一律回落
|
||||
// 默认额 —— 「不限流」不该是敲一个 0 就能得到的状态(design/12 §N C.5 G3
|
||||
// 登记过那条静默失败形态)。要完全不花模型额度,请关 `t1Judge`。
|
||||
T1PerWindow int `json:"t1PerWindow"`
|
||||
// T1WindowSec 是上面那个额度的时间窗(**秒**)。零/负 = 没配 ⇒ 用代码默认值
|
||||
// (memory 侧的 defaultT1Window)。
|
||||
//
|
||||
// 用秒而不是 Go 的时长字面量(`"1h"`)是刻意的:这个文件是给人改的,写
|
||||
// `3600` 比要求主人记住 Go 的时长语法更不容易写错,也与既有
|
||||
// `$DATA_DIR/config/backthink.json` 的 `*_SEC` 口径一致。
|
||||
T1WindowSec int `json:"t1WindowSec"`
|
||||
}
|
||||
|
||||
// AutoApplyOff 报告自动档是否被这份配置**显式关掉**。nil(没配)返回 false ——
|
||||
|
||||
@@ -134,6 +134,54 @@ func TestLoadMemoryTidyTuning_IngestGate(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoadMemoryTidyTuning_T1RateLimit 钉住 **T1 全局限速**的两个数值键
|
||||
// (design/23 §15.4,主人 2026-10-07 裁定):它们与 `t1Judge` 同一个文件、
|
||||
// **零新 env 键**,且**零值只能是「没配」**——绝不能是「无上限」。
|
||||
func TestLoadMemoryTidyTuning_T1RateLimit(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// ① 没写 ⇒ 保持零值(调用方据此回落代码默认额,不是「不限流」)。
|
||||
absent := writeMemoryTidyFile(t, `{"t1Judge": true}`)
|
||||
got, err := LoadMemoryTidyTuning(absent)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadMemoryTidyTuning: %v", err)
|
||||
}
|
||||
if got.T1PerWindow != 0 || got.T1WindowSec != 0 {
|
||||
t.Fatalf("没写限速键应当给零值:%+v", got)
|
||||
}
|
||||
|
||||
// ② 写了 ⇒ 如实回读,且不连带改动别的键(一个键不能顺手关掉另一件事)。
|
||||
set := writeMemoryTidyFile(t, `{"t1PerWindow": 240, "t1WindowSec": 1800}`)
|
||||
got, err = LoadMemoryTidyTuning(set)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadMemoryTidyTuning(set): %v", err)
|
||||
}
|
||||
if got.T1PerWindow != 240 || got.T1WindowSec != 1800 {
|
||||
t.Fatalf("限速键回读 = %d/%d, want 240/1800", got.T1PerWindow, got.T1WindowSec)
|
||||
}
|
||||
if got.T1JudgeOff() || got.IngestGateOff() || got.AutoApplyOff() {
|
||||
t.Fatalf("限速键顺手改了别的开关:%+v", got)
|
||||
}
|
||||
|
||||
// ③ 显式 0 ⇒ 仍是零值(「没法用 0 关掉限流」是刻意的:那会是一处静默的失效面)。
|
||||
zero := writeMemoryTidyFile(t, `{"t1PerWindow": 0, "t1WindowSec": 0}`)
|
||||
got, err = LoadMemoryTidyTuning(zero)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadMemoryTidyTuning(zero): %v", err)
|
||||
}
|
||||
if got.T1PerWindow != 0 || got.T1WindowSec != 0 {
|
||||
t.Fatalf("显式 0 的回读 = %+v", got)
|
||||
}
|
||||
|
||||
// ④ 类型错 ⇒ 分类错误(调用方据此记 WARN 并回落代码默认额,绝不静默不限流)。
|
||||
for _, body := range []string{`{"t1PerWindow": "many"}`, `{"t1WindowSec": "1h"}`} {
|
||||
broken := writeMemoryTidyFile(t, body)
|
||||
if _, err := LoadMemoryTidyTuning(broken); !errors.Is(err, ErrInvalidMemoryTidy) {
|
||||
t.Fatalf("坏配置 %s 的 err = %v, want ErrInvalidMemoryTidy", body, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDefaultMemoryTidyPathIsDataDirConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
got := DefaultMemoryTidyPath(filepath.Join("root", "data"))
|
||||
|
||||
@@ -31,7 +31,9 @@ import (
|
||||
// 3. **规则优先、模型兜底**:T0 是纯规则(design §15 D3-2 T0),**恒为零模型
|
||||
// 调用**;T1(灰区调一次轻量模型,§15.1 D3-2)默认**开**,但只在「T0 判不出
|
||||
// 结论且存在灰区候选」时才发出**至多一次**调用(preinsert_t1.go),并且
|
||||
// 要有模型出口才成立 —— 两者任一缺席即是结构性零模型调用。
|
||||
// 要有模型出口才成立 —— 两者任一缺席即是结构性零模型调用。每一次这样的
|
||||
// 调用还要先过**全局限速**(默认 120 次/小时,preinsert_t1_limit.go):
|
||||
// 超限即跳过 T1、回落 create + `t1_throttled`。
|
||||
// **同步入库闸(preinsert_gate.go)不碰 T1**,写入路径因此仍是零模型调用。
|
||||
//
|
||||
// 代价(如实登记):极短窗口内库里会多一行待裁定的重复条目,随后被软合并
|
||||
@@ -141,6 +143,11 @@ type PreInsertReviewer struct {
|
||||
log *slog.Logger
|
||||
now func() time.Time
|
||||
queue chan MemoryItem
|
||||
// t1Gate 是 T1 的**全局限速**闸(design/23 §15.4)。挂在实例上而不是包级
|
||||
// 变量:AGENTS §10 禁包级全局变量,而组合根只建一个审查服务 ⇒ 实例字段
|
||||
// 事实上就是「全进程共享一个窗口额度」。它与 r.now 同源,因此测试可以整体
|
||||
// 推进时钟(见 preinsert_t1_limit.go)。
|
||||
t1Gate *t1RateGate
|
||||
|
||||
mu sync.Mutex
|
||||
running bool
|
||||
@@ -168,6 +175,11 @@ func NewPreInsertReviewer(cfg ReviewConfig, deps ReviewDeps) (*PreInsertReviewer
|
||||
log: logger.WithSource(baseTidyLog(deps.Log), "memory/review"),
|
||||
now: now,
|
||||
queue: make(chan MemoryItem, cfg.QueueSize),
|
||||
// 限流闸在 normalize **之后**构建:闸把非正的额度读成「恒放行」,因此
|
||||
// 归一前的 ≤0 一旦直达它,「回落默认」就会静默变成「无上限」
|
||||
// (design/12 §N C.5 G3 登记过的失败形态)。这里的 cfg 来自上面那行
|
||||
// normalizeReviewConfig。
|
||||
t1Gate: newT1RateGate(cfg.T1PerWindow, cfg.T1Window, now),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -243,6 +255,10 @@ func (r *PreInsertReviewer) Start(ctx context.Context) {
|
||||
"t1_off_by_config", r.cfg.DisableT1,
|
||||
"t1_timeout", r.t1Timeout().String(),
|
||||
"t1_max_gray", defaultT1MaxGray,
|
||||
// T1 的全局限速额度(已 normalize 的**生效值**,不是配置原文):
|
||||
// 「以为不限流其实限着」与反过来的误解都在启动行上即可排除。
|
||||
"t1_per_window", r.cfg.T1PerWindow,
|
||||
"t1_window", r.cfg.T1Window.String(),
|
||||
"sweep_interval", DefaultIngestReviewGCInterval.String(),
|
||||
"retention", DefaultIngestReviewRetention.String(),
|
||||
"max_reviews", DefaultIngestReviewMaxRows)
|
||||
|
||||
@@ -24,6 +24,18 @@ const (
|
||||
// 不是任何角色(design §4.1 C 的同一理由)。
|
||||
reviewAuditWho = "system/memoryreview"
|
||||
reviewAuditVerb = "memory.review.ingest"
|
||||
|
||||
// defaultT1PerWindow / defaultT1Window 是 T1 **全局限速**的默认额度
|
||||
// (design/23 §15.4,主人 2026-10-07 裁定「加一层保守的全局速率限流」)。
|
||||
//
|
||||
// 推算依据(2026-10-07):现网节奏是每天 700–1000 条新记忆(≈25–42 条/小时),
|
||||
// 而 T1 只在「T0 判不出结论**且**存在灰区候选」时才发调用 —— 每条新记忆至多换
|
||||
// 一次调用。取 **120 次 / 60 分钟**,即在最忙的那一小时也留 ≈3–5 倍余量:
|
||||
// 正常节奏下这个闸几乎碰不到;只有「召回/规则突然把大批条目判成灰区」这类
|
||||
// 异常形态才会触顶,而那正是要拦的东西。超限不是错误:跳过 T1、回落 create、
|
||||
// 留痕 `t1_throttled`,记忆一条不丢、一个字不改。
|
||||
defaultT1PerWindow = 120
|
||||
defaultT1Window = 60 * time.Minute
|
||||
)
|
||||
|
||||
// ReviewConfig 携带入库前审查的调优旋钮。零值经 normalize 选取安全默认值。
|
||||
@@ -73,6 +85,20 @@ type ReviewConfig struct {
|
||||
// T1Timeout 是**单次** T1 模型调用的墙钟上限。非正用代码默认值
|
||||
// defaultT1Timeout;最终还会被 cfg.Timeout 收紧取小(见 t1Timeout)。
|
||||
T1Timeout time.Duration
|
||||
// T1PerWindow 是 T1 的**全局**窗口额度(次)—— 不是每条各自的限制,而是
|
||||
// 全进程共享的一个窗口:T1Window 内至多这么多次模型调用,超限即跳过 T1 并
|
||||
// 回落 `create`(reason=`t1_throttled`,design/23 §15.4)。
|
||||
//
|
||||
// 非正用代码默认值 defaultT1PerWindow。**刻意没有「无上限」语义**:
|
||||
// design/12 §N C.5 G3 登记过那条静默失败形态(未归一的 ≤0 直达滑窗 ⇒ 被读成
|
||||
// 「无限制」,没有报错、没有 WARN);要完全不花钱请用
|
||||
// `$DATA_DIR/config/memory-tidy.json` 的 `{"t1Judge": false}` 关掉 T1。
|
||||
//
|
||||
// 额度**只被真正的模型调用消耗**(闸放行才记账,被拒的那次不占),因此
|
||||
// 「窗口过了就恢复」是结构性成立的。
|
||||
T1PerWindow int
|
||||
// T1Window 是 T1PerWindow 的时间窗。非正用代码默认值 defaultT1Window。
|
||||
T1Window time.Duration
|
||||
}
|
||||
|
||||
// normalizeReviewConfig 用设计稿默认值填充零值字段。
|
||||
@@ -98,5 +124,14 @@ func normalizeReviewConfig(cfg ReviewConfig) ReviewConfig {
|
||||
if cfg.T1Timeout <= 0 {
|
||||
cfg.T1Timeout = defaultT1Timeout
|
||||
}
|
||||
// 🔴 限流的两项必须在这里回落(design/12 §N C.5 G3 的教训):闸是照着
|
||||
// 「非正 = 恒放行」实现的,未归一的 ≤0 直达它就会把「回落默认」**静默**变成
|
||||
// 「无上限」。因此 NewPreInsertReviewer 保证闸在 normalize **之后**构建。
|
||||
if cfg.T1PerWindow <= 0 {
|
||||
cfg.T1PerWindow = defaultT1PerWindow
|
||||
}
|
||||
if cfg.T1Window <= 0 {
|
||||
cfg.T1Window = defaultT1Window
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
@@ -33,6 +33,11 @@ type ReviewStats struct {
|
||||
// 的差值就是「调了但没用上」(非法回复 / 超时 / 报错回落 create)——
|
||||
// 这个差值本身是可观测信号,因此两个数分开计。
|
||||
T1Decided int
|
||||
// T1Throttled 是**全局限速**命中、因而跳过 T1 的次数(design/23 §15.4,
|
||||
// reason=`t1_throttled`)。它**不**计入 T1Called(一次模型调用都没发),
|
||||
// 也不计入 T1Decided(没定任何结论)——「想调但被闸拦下」是与「调了没用上」
|
||||
// 不同的一件事,分开计才看得出额度是不是成了瓶颈。
|
||||
T1Throttled int
|
||||
}
|
||||
|
||||
// bump 在互斥下修改计数(工作协程、手动 Review 与同步闸都可能调用)。
|
||||
@@ -75,7 +80,7 @@ func (r *PreInsertReviewer) auditWith(verb string, rev IngestReview) {
|
||||
// reviewStatsLabel 把计数渲染成一条稳定日志串。
|
||||
func reviewStatsLabel(s ReviewStats) string {
|
||||
return fmt.Sprintf("reviewed=%d skip=%d merge=%d modify=%d create=%d dropped=%d failed=%d "+
|
||||
"gate_checked=%d gate_absorbed=%d gate_failed=%d t1_called=%d t1_decided=%d",
|
||||
"gate_checked=%d gate_absorbed=%d gate_failed=%d t1_called=%d t1_decided=%d t1_throttled=%d",
|
||||
s.Reviewed, s.Skipped, s.Merged, s.Modified, s.Created, s.Dropped, s.Failed,
|
||||
s.GateChecked, s.GateAbsorbed, s.GateFailed, s.T1Called, s.T1Decided)
|
||||
s.GateChecked, s.GateAbsorbed, s.GateFailed, s.T1Called, s.T1Decided, s.T1Throttled)
|
||||
}
|
||||
@@ -34,6 +34,10 @@ import (
|
||||
// 4. **有界 + 不阻塞写入**:每次审查**最多一次**模型调用(灰区候选打包进同一个
|
||||
// prompt,最多 `defaultT1MaxGray` 条),且有自己的子超时(≤ 单条审查总预算)。
|
||||
// 超时 / 报错 / 非法回复 ⇒ `create` + WARN + **零改写**,绝不阻塞、绝不丢记忆。
|
||||
// 此外还有一层**全局滑窗限速**(`defaultT1PerWindow` 次 / `defaultT1Window`,
|
||||
// 全进程共享一个窗口)堵住最坏情形:现网每天数百到上千条新记忆,若大半带灰区
|
||||
// 候选就是每天数百到上千次调用;超限 ⇒ 跳过 T1 + 回落 `create` +
|
||||
// `t1_throttled`(见 preinsert_t1_limit.go)。
|
||||
// T1 只挂在**异步审查层**(`PreInsertReviewer.Review`);同步入库闸
|
||||
// (preinsert_gate.go)不碰它 ⇒ 写入路径仍是零模型调用。
|
||||
// 5. **不新造 LLM 客户端**:注入的是**既有** `model.Adapter` 的调用口(组合根
|
||||
@@ -41,10 +45,10 @@ import (
|
||||
// 厂商协议,也不新增依赖。
|
||||
//
|
||||
// 可观测(§15 D3-8):T1 是否参与、结论是什么,在留痕 `reason` 列上如实可辨
|
||||
// (`t1_merge` / `t1_modify` / `t1_keep` / `t1_reply_invalid` / `t1_unavailable`),
|
||||
// 规则版本随之升到 `IngestRuleVersion`。模型的自由文本理由**既不落库也不落日志**:
|
||||
// 留痕表按设计不存正文,而模型的理由可能改述正文 ⇒ 一律丢弃(它在 prompt 里的
|
||||
// 唯一作用,是逼模型先表态,从而把动作判得更稳)。
|
||||
// (`t1_merge` / `t1_modify` / `t1_keep` / `t1_reply_invalid` / `t1_unavailable` /
|
||||
// `t1_throttled`),规则版本随之升到 `IngestRuleVersion`。模型的自由文本理由
|
||||
// **既不落库也不落日志**:留痕表按设计不存正文,而模型的理由可能改述正文 ⇒ 一律
|
||||
// 丢弃(它在 prompt 里的唯一作用,是逼模型先表态,从而把动作判得更稳)。
|
||||
|
||||
const (
|
||||
// defaultT1MaxGray 是打包进**一次** prompt 的灰区候选上限。灰区候选按召回
|
||||
@@ -71,6 +75,11 @@ const (
|
||||
t1ReasonKeep = "t1_keep"
|
||||
t1ReasonReplyInvalid = "t1_reply_invalid"
|
||||
t1ReasonUnavailable = "t1_unavailable"
|
||||
// t1ReasonThrottled 是**全局限速**命中时的原因码(preinsert_t1_limit.go,
|
||||
// design/23 §15.4)。它与上面两个失败码同族、同样只改 `reason`:决定恒
|
||||
// `create`、零改写。与它们的唯一区别是语义 —— 失败是「模型没答上」,
|
||||
// 限流是「这一轮先不判、留给以后」,因此不走失败那条 WARN。
|
||||
t1ReasonThrottled = "t1_throttled"
|
||||
)
|
||||
|
||||
// ErrT1ReplyInvalid 是模型回复解不出三态枚举(或候选编号越界)时的分类错误。
|
||||
@@ -118,12 +127,19 @@ func (r *PreInsertReviewer) t1Timeout() time.Duration {
|
||||
// maybeJudgeT1 在「T0 判不出结论 + 有灰区候选」时,用**一次**模型调用改判。
|
||||
//
|
||||
// 任何非「模型给出合法三态」的情形都**原样返回 v**:未启用、没有模型出口、
|
||||
// 没有灰区候选、调用超时/报错、回复非法。因此「失败 ⇒ create + 零改写」是
|
||||
// 结构性成立的 —— 这条路径上根本没有改写的代码。
|
||||
// 没有灰区候选、**全局额度用尽(限流)**、调用超时/报错、回复非法。因此
|
||||
// 「失败/限流 ⇒ create + 零改写」是结构性成立的 —— 这条路径上根本没有改写的
|
||||
// 代码。
|
||||
func (r *PreInsertReviewer) maybeJudgeT1(ctx context.Context, item MemoryItem, v ingestVerdict) ingestVerdict {
|
||||
if !r.t1Enabled() || len(v.Gray) == 0 {
|
||||
return v
|
||||
}
|
||||
// 全局限速(design/23 §15.4)摆在**发请求之前**,且排在两个前置守卫**之后**:
|
||||
// 没有灰区候选的审查一次额度都不占(它本来就不调模型),只有真的要花钱时才
|
||||
// 过闸、才记账 —— 于是「超限」永远等于「本来会多花一次调用」。
|
||||
if allowed, firstRefusal := r.t1Gate.allow(); !allowed {
|
||||
return r.t1Throttled(item, v, firstRefusal)
|
||||
}
|
||||
gray := v.Gray
|
||||
if len(gray) > defaultT1MaxGray {
|
||||
gray = gray[:defaultT1MaxGray]
|
||||
@@ -149,6 +165,9 @@ func (r *PreInsertReviewer) maybeJudgeT1(ctx context.Context, item MemoryItem, v
|
||||
// callT1 发起**唯一**的那次 T1 调用(每次审查至多一次)。它自带子超时,并把
|
||||
// 同一个预算也交给适配器(model.Options.Timeout)—— 进程内与 HTTP 侧各把一道,
|
||||
// 一次灰区判定绝不可能拖长审查。模型正文(prompt 与响应)**不进日志**。
|
||||
//
|
||||
// 走到这里说明额度**已经**被闸放行(放行才记账):因此「这一条发得出去」与
|
||||
// 「这条占掉了一格额度」是同一件事,不存在「调了但没记账」的缝。
|
||||
func (r *PreInsertReviewer) callT1(ctx context.Context, item MemoryItem, gray []ingestCandidate) (t1Reply, error) {
|
||||
budget := r.t1Timeout()
|
||||
cctx, cancel := context.WithTimeout(ctx, budget)
|
||||
@@ -178,6 +197,30 @@ func (r *PreInsertReviewer) t1Fallback(ctx context.Context, item MemoryItem, v i
|
||||
return v
|
||||
}
|
||||
|
||||
// t1Throttled 组装**限流**回落结论:与 t1Fallback 逐字同结果 —— 决定恒 `create`、
|
||||
// 只把 `reason` 换成 `t1_throttled`、**零改写**,差分只在外面的口径与日志档位。
|
||||
//
|
||||
// 为什么日志与失败路径不同:失败(超时 / 报错 / 回复非法)是**故障**,每条都该
|
||||
// 现形;超限是**预期内的省钱行为**(主人 2026-10-07:「超限不是错误,是这一轮先
|
||||
// 不判、留给以后」),一天可能几百条 —— 全记 WARN 会把真正的故障淹掉。因此:
|
||||
// 每个窗口只记**一条** WARN(附窗口额度与窗口长度,够运维定位),其余记 Debug。
|
||||
// 留痕 `reason` 一律如实(那才是机器可查的证据面)。日志只带 id / 作用域 / 额度,
|
||||
// **绝不带记忆正文**。
|
||||
func (r *PreInsertReviewer) t1Throttled(item MemoryItem, v ingestVerdict, firstRefusal bool) ingestVerdict {
|
||||
v.Reason = t1ReasonThrottled
|
||||
r.bump(func(s *ReviewStats) { s.T1Throttled++ })
|
||||
args := []any{
|
||||
"item_id", item.ID, "scope", string(item.Scope), "reason", t1ReasonThrottled,
|
||||
"t1_per_window", r.cfg.T1PerWindow, "t1_window", r.cfg.T1Window.String(),
|
||||
}
|
||||
if firstRefusal {
|
||||
r.log.Warn("memory ingest review t1 judge throttled, gray-zone item kept as create", args...)
|
||||
return v
|
||||
}
|
||||
r.log.Debug("memory ingest review t1 judge throttled, gray-zone item kept as create", args...)
|
||||
return v
|
||||
}
|
||||
|
||||
// t1Verdict 把模型回复映射成判定结论。ok=false 表示回复不可用(非法枚举 /
|
||||
// 编号越界),调用方据此回落 create —— **绝不猜、绝不挑一个最近的候选**。
|
||||
func t1Verdict(reply t1Reply, item MemoryItem, gray []ingestCandidate, base ingestVerdict) (ingestVerdict, bool) {
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 本文件是 T1(灰区模型判定)的**全局限速闸**(design/23 §15.4 的「加一层保守的
|
||||
// 全局速率限流」,主人 2026-10-07 裁定)。
|
||||
//
|
||||
// 为什么要有它:T1 默认**开**,而它只在「T0 判不出结论 + 存在灰区候选」时才发一次
|
||||
// 模型调用 —— 现网节奏是每天 700–1000 条新记忆,若其中多数都带灰区候选,就是
|
||||
// **每天数百到上千次**额外调用。在此之前 T1 上**没有任何上限**:那是一处没有刹车
|
||||
// 的额度出口。加的这层闸是**全进程共享的一个窗口额度**(不是每条各自的限制):
|
||||
// 超限 ⇒ 跳过 T1 并回落 `create`(reason=`t1_throttled`),与既有失败路径逐字同
|
||||
// 语义(**只改 reason、决定恒 create、零改写**)。超限**不是错误**,是「这一轮
|
||||
// 先不判、留给以后」—— 记忆一条不丢、一个字不改。
|
||||
//
|
||||
// 为什么自研这一件、而不是复用既有滑窗件 —— 两条都是硬理由(不是图省事):
|
||||
//
|
||||
// 1. **窗口过期必须可确定性测试**。既有共享件是 `session.NewRateLimiter`
|
||||
// (`internal/channel/session/proactive.go`,design/12 §N C.5 指定的那个),
|
||||
// 它内部写死 `time.Now()`,**没有时钟注入缝**:它的「窗口过后额度恢复」只能
|
||||
// 靠「极小窗口 + 真实短等」覆盖 —— 那正是 design/12 §N C.5 自己登记的已知
|
||||
// 取舍。本切片要求用**注入时钟**验这条通路(与本包 `ReviewDeps.Now` 同一
|
||||
// 手法),而 `allow()` 的每一次判定都要看时间,没有缝就做不到。
|
||||
// 2. **依赖方向**:那一件住在 `internal/channel/session`,且会带进
|
||||
// `internal/channel/adapter`(渠道层的适配器/对话概念)。记忆运行时为 20 行的
|
||||
// 计数去依赖渠道层,是拿架构换省事。
|
||||
//
|
||||
// design/12 §N C.5 对同一取舍已经写下了结论:「若主会话要求完全确定,改法是让闸
|
||||
// 自带 `now func() time.Time`(约 +15 行,代价是出现第二份滑窗实现)」。本文件
|
||||
// 就是那一改法的落地,并**刻意只做这一件事**:它不认识令牌桶、不按 key 分桶,
|
||||
// 只有一个窗口一份额度 —— 不是「第二套通用限流机制」,是 T1 的一处刹车。
|
||||
|
||||
// t1RateGate 是 T1 的全局限速闸:一个滑动窗口额度,**全进程共享一个实例**
|
||||
// (组合根只建一个 `PreInsertReviewer`,闸就挂在那一个实例上 ⇒ 事实上是全进程
|
||||
// 额度;AGENTS §10 禁包级全局变量,因此它只能是实例字段)。
|
||||
//
|
||||
// 并发安全:审查可能来自后台工作协程,也可能来自测试/手动 Review,状态全部在
|
||||
// 互斥下读写。
|
||||
type t1RateGate struct {
|
||||
limit int
|
||||
window time.Duration
|
||||
now func() time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
hits []time.Time
|
||||
// refused 记账「本窗口已经拒过一次」,只为把日志收敛成**每窗口一条 WARN**:
|
||||
// 超限是预期内的省钱行为、不是故障,一天几百条 WARN 会把真正的故障淹掉。
|
||||
refused bool
|
||||
}
|
||||
|
||||
// newT1RateGate 构建闸。limit / window **必须是 normalize 之后的正值**:非正会
|
||||
// 得到一个恒放行的闸(见 allow),那正是 design/12 §N C.5 G3 登记过的静默失败
|
||||
// 形态「≤0 被读成无上限」—— 因此调用方(NewPreInsertReviewer)保证它在
|
||||
// `normalizeReviewConfig` **之后**构建,绝不让未归一的 ≤0 直达滑窗。
|
||||
func newT1RateGate(limit int, window time.Duration, now func() time.Time) *t1RateGate {
|
||||
if now == nil {
|
||||
now = time.Now
|
||||
}
|
||||
return &t1RateGate{limit: limit, window: window, now: now}
|
||||
}
|
||||
|
||||
// allow 报告这一次 T1 调用是否放行;第二个返回值是「本窗口的第一次拒绝」,
|
||||
// 调用方据此把日志收成每窗口一条 WARN、其余记 Debug。
|
||||
//
|
||||
// 额度口径与既有滑窗件一致:**放行才消耗** —— 被拒的那次不记账,因此超限不会把
|
||||
// 窗口往后推,额度到点(最后一次放行的窗口之后)自然恢复。
|
||||
//
|
||||
// nil 或非正额度 ⇒ 恒放行:那是「没配限流器」的安全默认,而不是「静默关掉一处
|
||||
// 防护」(构造处已由 normalize 保证不会发生)。
|
||||
func (g *t1RateGate) allow() (allowed bool, firstRefusal bool) {
|
||||
if g == nil || g.limit <= 0 || g.window <= 0 {
|
||||
return true, false
|
||||
}
|
||||
now := g.now()
|
||||
cutoff := now.Add(-g.window)
|
||||
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
|
||||
keep := 0
|
||||
for _, t := range g.hits {
|
||||
if t.After(cutoff) {
|
||||
g.hits[keep] = t
|
||||
keep++
|
||||
}
|
||||
}
|
||||
g.hits = g.hits[:keep]
|
||||
if keep == 0 {
|
||||
// 窗口已排空 ⇒ 新一轮开始,「本窗口已报过一次 WARN」随之复位。
|
||||
g.refused = false
|
||||
}
|
||||
if keep >= g.limit {
|
||||
first := !g.refused
|
||||
g.refused = true
|
||||
return false, first
|
||||
}
|
||||
g.hits = append(g.hits, now)
|
||||
return true, false
|
||||
}
|
||||
@@ -0,0 +1,412 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
|
||||
)
|
||||
|
||||
// 本文件是 **T1 全局限速**(design/23 §15.4,preinsert_t1_limit.go)的测试:
|
||||
// 超限即不调模型、额度只被真调用消耗、窗口过后恢复、限流路径零改写、日志收敛
|
||||
// (每窗口一条 WARN),以及「≤0 不许被读成无上限」那条归一化红线。
|
||||
//
|
||||
// 三态落地 / 失败回落 / 开关口径在 preinsert_t1_test.go,调用次数有界与硬门隔离
|
||||
// 在 preinsert_t1_bound_test.go(本文件不动它们,只加限流这一层)。
|
||||
//
|
||||
// 所有用例都**不**靠真实等待:窗口过期由可控时钟推进(下面 t1Clock),与
|
||||
// t1JudgeStub「被调用即 panic」是同一思路 —— 把「不该发生的调用」变成立刻现形。
|
||||
|
||||
// t1Clock 是可推进的墙钟:限流的「窗口过后额度恢复」必须由它驱动,真实等待会让
|
||||
// 测试变慢且不确定(窗口是墙钟量,不是逻辑量)。
|
||||
type t1Clock struct {
|
||||
mu sync.Mutex
|
||||
at time.Time
|
||||
}
|
||||
|
||||
func newT1Clock(at time.Time) *t1Clock { return &t1Clock{at: at} }
|
||||
|
||||
func (c *t1Clock) now() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.at
|
||||
}
|
||||
|
||||
func (c *t1Clock) advance(d time.Duration) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.at = c.at.Add(d)
|
||||
}
|
||||
|
||||
// newT1LimiterReviewer 构建「异步审查层 + 已接线 T1 + 可控时钟」的服务:
|
||||
// 同步入库闸显式关掉(这批用例只验异步那一层,否则测到的是闸),其余走默认值;
|
||||
// 限流额度由 cfg.T1PerWindow / cfg.T1Window 给出(非正即回落代码默认)。
|
||||
func newT1LimiterReviewer(
|
||||
t *testing.T, store *Store, emb embedding.Embedder, cfg ReviewConfig,
|
||||
judge T1Judge, now func() time.Time, log *slog.Logger,
|
||||
) *PreInsertReviewer {
|
||||
t.Helper()
|
||||
cfg.Enabled = true
|
||||
if cfg.QueueSize == 0 {
|
||||
cfg.QueueSize = 8
|
||||
}
|
||||
if cfg.ScanLimit == 0 {
|
||||
cfg.ScanLimit = 50
|
||||
}
|
||||
if cfg.Timeout == 0 {
|
||||
cfg.Timeout = 5 * time.Second
|
||||
}
|
||||
if cfg.Threshold == 0 {
|
||||
cfg.Threshold = 0.80
|
||||
}
|
||||
cfg.DisableIngestGate = true
|
||||
if log == nil {
|
||||
log = testLogger()
|
||||
}
|
||||
r, err := NewPreInsertReviewer(cfg, ReviewDeps{
|
||||
Store: store, Embedder: emb, Judge: judge, Now: now, Log: log,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewPreInsertReviewer: %v", err)
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// TestT1ThrottledSkipsModelCallAndFallsBackToCreate 是任务书 §5①:额度用尽后
|
||||
// **不再发起模型调用**,且结论仍是 `create` + `reason=t1_throttled`(与既有失败
|
||||
// 路径同语义:只改 reason、决定恒 create、零改写)。
|
||||
//
|
||||
// 桩回 `keep` 是刻意的:灰区条目因此每次都留在册,同一条可以被反复审 —— 于是
|
||||
// 「第 N 次」这件事只由额度决定,与判定结果无关。
|
||||
func TestT1ThrottledSkipsModelCallAndFallsBackToCreate(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
_, fresh := seedT1GrayPair(t, store)
|
||||
|
||||
judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`}
|
||||
r := newT1LimiterReviewer(t, store, t1GrayEmbedder(),
|
||||
ReviewConfig{T1PerWindow: 2, T1Window: time.Hour}, judge.judge, nil, nil)
|
||||
|
||||
for i := 1; i <= 2; i++ {
|
||||
rev, err := r.Review(ctx, fresh)
|
||||
if err != nil {
|
||||
t.Fatalf("第 %d 次 Review: %v", i, err)
|
||||
}
|
||||
if rev.Reason != t1ReasonKeep {
|
||||
t.Fatalf("额度内第 %d 次的 reason = %q, want %q", i, rev.Reason, t1ReasonKeep)
|
||||
}
|
||||
}
|
||||
if judge.count() != 2 {
|
||||
t.Fatalf("额度内模型调用 = %d 次, want 2", judge.count())
|
||||
}
|
||||
|
||||
rev, err := r.Review(ctx, fresh)
|
||||
if err != nil {
|
||||
t.Fatalf("超限那次 Review: %v", err)
|
||||
}
|
||||
if judge.count() != 2 {
|
||||
t.Fatalf("超限后仍调了模型:共 %d 次, want 仍为 2", judge.count())
|
||||
}
|
||||
if rev.Decision != IngestCreate {
|
||||
t.Fatalf("超限后决定 = %q, want create(限流不是报错,是这一轮先不判)", rev.Decision)
|
||||
}
|
||||
if rev.Reason != t1ReasonThrottled {
|
||||
t.Fatalf("超限后 reason = %q, want %q", rev.Reason, t1ReasonThrottled)
|
||||
}
|
||||
st := r.Stats()
|
||||
if st.T1Called != 2 || st.T1Decided != 2 || st.T1Throttled != 1 {
|
||||
t.Fatalf("计数 = %+v, want T1Called=2 T1Decided=2 T1Throttled=1", st)
|
||||
}
|
||||
if label := reviewStatsLabel(st); !strings.Contains(label, "t1_throttled=1") {
|
||||
t.Fatalf("统计串里看不到限流:%q", label)
|
||||
}
|
||||
}
|
||||
|
||||
// TestT1ThrottleRecoversAfterWindow 是任务书 §5②:额度在窗口过后**恢复**
|
||||
// (限流不把窗口往后推)。时钟由注入的 t1Clock 推进,不靠真实等待。
|
||||
func TestT1ThrottleRecoversAfterWindow(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
_, fresh := seedT1GrayPair(t, store)
|
||||
|
||||
clock := newT1Clock(tidyTestTime(t))
|
||||
judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`}
|
||||
window := 30 * time.Minute
|
||||
r := newT1LimiterReviewer(t, store, t1GrayEmbedder(),
|
||||
ReviewConfig{T1PerWindow: 1, T1Window: window}, judge.judge, clock.now, nil)
|
||||
|
||||
if rev, err := r.Review(ctx, fresh); err != nil || rev.Reason != t1ReasonKeep {
|
||||
t.Fatalf("窗口内首次 = %+v err=%v, want keep", rev, err)
|
||||
}
|
||||
if rev, err := r.Review(ctx, fresh); err != nil || rev.Reason != t1ReasonThrottled {
|
||||
t.Fatalf("窗口内第二次 = %+v err=%v, want %q", rev, err, t1ReasonThrottled)
|
||||
}
|
||||
|
||||
// 差一秒钟到期:仍然超限(边界口径与既有滑窗件逐字相同:严格晚于 cutoff 才算在窗内)。
|
||||
clock.advance(window - time.Second)
|
||||
if rev, err := r.Review(ctx, fresh); err != nil || rev.Reason != t1ReasonThrottled {
|
||||
t.Fatalf("窗口未满却恢复了额度:%+v err=%v", rev, err)
|
||||
}
|
||||
if judge.count() != 1 {
|
||||
t.Fatalf("窗口未满时模型调用 = %d 次, want 1", judge.count())
|
||||
}
|
||||
|
||||
// 推进过窗口:额度恢复,模型再次被调用。
|
||||
clock.advance(2 * time.Second)
|
||||
rev, err := r.Review(ctx, fresh)
|
||||
if err != nil {
|
||||
t.Fatalf("窗口过后 Review: %v", err)
|
||||
}
|
||||
if judge.count() != 2 {
|
||||
t.Fatalf("窗口过后模型调用 = %d 次, want 2(额度必须恢复)", judge.count())
|
||||
}
|
||||
if rev.Reason != t1ReasonKeep {
|
||||
t.Fatalf("窗口过后的 reason = %q, want %q", rev.Reason, t1ReasonKeep)
|
||||
}
|
||||
if st := r.Stats(); st.T1Throttled != 2 {
|
||||
t.Fatalf("超限计数 = %d, want 2", st.T1Throttled)
|
||||
}
|
||||
}
|
||||
|
||||
// TestT1ThrottledNeverTouchesMemory 是任务书 §5③:限流路径**只改 reason**,
|
||||
// 记忆行一个字不改(与 TestReviewFailureNeverTouchesMemory 同一手法)。
|
||||
//
|
||||
// 守卫手法照 TestT1DisabledMakesZeroModelCalls:把桩换成「被调用即 panic」并回
|
||||
// `merge` —— 限流一旦失效,要么当场 panic,要么真把两行改写掉,两种都在断言里现形。
|
||||
func TestT1ThrottledNeverTouchesMemory(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
existing, fresh := seedT1GrayPair(t, store)
|
||||
|
||||
judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`}
|
||||
r := newT1LimiterReviewer(t, store, t1GrayEmbedder(),
|
||||
ReviewConfig{T1PerWindow: 1, T1Window: time.Hour}, judge.judge, nil, nil)
|
||||
|
||||
// 第一次:额度内,桩回 keep(不改写任何行)。
|
||||
if _, err := r.Review(ctx, fresh); err != nil {
|
||||
t.Fatalf("额度内 Review: %v", err)
|
||||
}
|
||||
if judge.count() != 1 {
|
||||
t.Fatalf("额度内模型调用 = %d 次, want 1", judge.count())
|
||||
}
|
||||
|
||||
// 第二次:额度已用尽。桩换成「被调用即 panic」并回 merge。
|
||||
judge.mu.Lock()
|
||||
judge.panics = true
|
||||
judge.reply = `{"action":"merge","target":1}`
|
||||
judge.mu.Unlock()
|
||||
|
||||
rev := t1ReviewGuarded(t, r, fresh)
|
||||
if rev.Decision != IngestCreate || rev.Reason != t1ReasonThrottled {
|
||||
t.Fatalf("超限那次 = %+v, want create + %q", rev, t1ReasonThrottled)
|
||||
}
|
||||
if got := readReviewRow(t, store, fresh.ID); got.status != tidyStatusActive ||
|
||||
got.mergedInto != "" || got.content != fixtureT1CodeViaChannel {
|
||||
t.Fatalf("限流却改写了新行:%+v", got)
|
||||
}
|
||||
old := readReviewRow(t, store, existing.ID)
|
||||
if old.status != tidyStatusActive || old.mergeCount != 1 || old.content != fixtureT1CodeAgreed {
|
||||
t.Fatalf("限流却改了既有行:%+v", old)
|
||||
}
|
||||
if n := countActiveRows(t, store, "persona.a"); n != 2 {
|
||||
t.Fatalf("活跃行数 = %d, want 2(限流是「不调」,不是「调失败」)", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestT1ThrottleQuotaIsOnlySpentByRealCalls 钉住「额度只被真调用消耗」:
|
||||
// 没有灰区候选的审查(本来就不调模型)一次额度都不占 —— 否则整库的普通新建
|
||||
// 会把 T1 的额度白白吃光,限流就成了随机停机。
|
||||
func TestT1ThrottleQuotaIsOnlySpentByRealCalls(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
base := tidyTestTime(t)
|
||||
_, fresh := seedT1GrayPair(t, store)
|
||||
// 一条与灰区对**毫不相似**的条目:它落不到任何候选上 ⇒ 零灰区、零调用。
|
||||
plain := seedTidyItem(t, store, habitItem(fixtureHabitVacation, base.Add(2*time.Hour), 0.9))
|
||||
|
||||
judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`}
|
||||
r := newT1LimiterReviewer(t, store, t1GrayEmbedder(),
|
||||
ReviewConfig{T1PerWindow: 1, T1Window: time.Hour}, judge.judge, nil, nil)
|
||||
|
||||
rev, err := r.Review(ctx, plain)
|
||||
if err != nil {
|
||||
t.Fatalf("零灰区那次 Review: %v", err)
|
||||
}
|
||||
if rev.Decision != IngestCreate || judge.count() != 0 {
|
||||
t.Fatalf("零灰区那次 = %+v calls=%d,不该调模型", rev, judge.count())
|
||||
}
|
||||
if st := r.Stats(); st.T1Throttled != 0 {
|
||||
t.Fatalf("零灰区那次被记成限流:%+v", st)
|
||||
}
|
||||
|
||||
// 额度仍然完好在册:真需要判定的这一条照样调得到模型。
|
||||
if rev, err := r.Review(ctx, fresh); err != nil || rev.Reason != t1ReasonKeep {
|
||||
t.Fatalf("灰区那条 = %+v err=%v, want keep(额度被普通新建吃掉了)", rev, err)
|
||||
}
|
||||
if judge.count() != 1 {
|
||||
t.Fatalf("模型调用 = %d 次, want 1", judge.count())
|
||||
}
|
||||
}
|
||||
|
||||
// TestT1ThrottleLogsOneWarnPerWindow 钉住日志口径:超限是**预期内的省钱行为**,
|
||||
// 不是故障 ⇒ 每窗口只记一条 WARN,其余记 Debug;且日志里绝不出现记忆正文。
|
||||
func TestT1ThrottleLogsOneWarnPerWindow(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
_, fresh := seedT1GrayPair(t, store)
|
||||
|
||||
var buf bytes.Buffer
|
||||
log := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))
|
||||
judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`}
|
||||
limit := ReviewConfig{T1PerWindow: 1, T1Window: time.Hour}
|
||||
r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), limit, judge.judge, nil, log)
|
||||
|
||||
// 三次审查、额度 1:第 1 次真调、第 2 次超限(本窗口第一条 WARN)、
|
||||
// 第 3 次仍超限(降级成 Debug)—— 日志条数 2、其中 WARN 1。
|
||||
for i := 0; i < 3; i++ {
|
||||
if _, err := r.Review(ctx, fresh); err != nil {
|
||||
t.Fatalf("第 %d 次 Review: %v", i, err)
|
||||
}
|
||||
}
|
||||
logged := buf.String()
|
||||
if n := strings.Count(logged, "t1 judge throttled"); n != 2 {
|
||||
t.Fatalf("限流日志条数 = %d, want 2(两次超限各一条):%q", n, logged)
|
||||
}
|
||||
if n := strings.Count(logged, "level=WARN"); n != 1 {
|
||||
t.Fatalf("限流 WARN 条数 = %d, want 1(每窗口一条):%q", n, logged)
|
||||
}
|
||||
for _, leaked := range []string{fixtureT1CodeAgreed, fixtureT1CodeViaChannel} {
|
||||
if strings.Contains(logged, leaked) {
|
||||
t.Fatalf("日志里泄漏了记忆正文 %q:%q", leaked, logged)
|
||||
}
|
||||
}
|
||||
|
||||
// 窗口排空之后「本窗口已报过」必须复位:新窗口的第一次超限仍然要 WARN。
|
||||
clock := newT1Clock(tidyTestTime(t))
|
||||
r2 := newT1LimiterReviewer(t, store, t1GrayEmbedder(), limit, judge.judge, clock.now, log)
|
||||
for i := 0; i < 2; i++ {
|
||||
if _, err := r2.Review(ctx, fresh); err != nil {
|
||||
t.Fatalf("新一轮第 %d 次 Review: %v", i, err)
|
||||
}
|
||||
}
|
||||
if n := strings.Count(buf.String(), "level=WARN"); n != 2 {
|
||||
t.Fatalf("新窗口的 WARN 条数 = %d, want 2:%q", n, buf.String())
|
||||
}
|
||||
clock.advance(2 * time.Hour)
|
||||
for i := 0; i < 2; i++ {
|
||||
if _, err := r2.Review(ctx, fresh); err != nil {
|
||||
t.Fatalf("过期轮第 %d 次 Review: %v", i, err)
|
||||
}
|
||||
}
|
||||
if n := strings.Count(buf.String(), "level=WARN"); n != 3 {
|
||||
t.Fatalf("跨窗口后的 WARN 条数 = %d, want 3(新窗口要重新报一次):%q", n, buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestT1DisabledLimitsNothingAndCallsNothing 是任务书 §5④:`t1Judge: false`
|
||||
// (DisableT1)时**零模型调用**,限流器连碰都不碰 —— 既有那条「被调用即 panic」
|
||||
// 的守卫测试(TestT1DisabledMakesZeroModelCalls)因此不受本切片影响。
|
||||
func TestT1DisabledLimitsNothingAndCallsNothing(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newTestStore(t)
|
||||
_, fresh := seedT1GrayPair(t, store)
|
||||
|
||||
judge := &t1JudgeStub{panics: true, reply: `{"action":"merge","target":1}`}
|
||||
r := newT1LimiterReviewer(t, store, t1GrayEmbedder(),
|
||||
ReviewConfig{DisableT1: true, T1PerWindow: 1, T1Window: time.Hour}, judge.judge, nil, nil)
|
||||
|
||||
rev := t1ReviewGuarded(t, r, fresh)
|
||||
if judge.count() != 0 {
|
||||
t.Fatalf("T1 关掉后仍调了 %d 次模型", judge.count())
|
||||
}
|
||||
if rev.Decision != IngestCreate || rev.Reason != "no_contained_match" {
|
||||
t.Fatalf("T1 关掉后 = %+v, want create + T0 原始原因码(限流不该改这里的语义)", rev)
|
||||
}
|
||||
if st := r.Stats(); st.T1Called != 0 || st.T1Decided != 0 || st.T1Throttled != 0 {
|
||||
t.Fatalf("T1 关掉后计数 = %+v, want 全零", st)
|
||||
}
|
||||
}
|
||||
|
||||
// TestT1RateGateIsConcurrencySafe 钉住闸的**并发安全**:审查可能来自后台工作
|
||||
// 协程,也可能来自手动 Review。N 个协程同时抢 K 个额度 ⇒ **恰好**放行 K 个 ——
|
||||
// 多放行就是超发模型调用(额度失效),少放行就是并发把额度吃掉(额度虚耗)。
|
||||
func TestT1RateGateIsConcurrencySafe(t *testing.T) {
|
||||
t.Parallel()
|
||||
const (
|
||||
limit = 7
|
||||
workers = 32
|
||||
)
|
||||
gate := newT1RateGate(limit, time.Hour, newT1Clock(tidyTestTime(t)).now)
|
||||
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
mu sync.Mutex
|
||||
allowed int
|
||||
)
|
||||
for i := 0; i < workers; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if ok, _ := gate.allow(); ok {
|
||||
mu.Lock()
|
||||
allowed++
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if allowed != limit {
|
||||
t.Fatalf("并发放行 = %d, want 恰为 %d", allowed, limit)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNormalizeReviewConfigFillsT1ThrottleDefaults 是 design/12 §N C.5 G3 那条
|
||||
// 红线的探针:额度**非正**必须被 normalize 回落成默认额 —— 绝不许带着 ≤0 进闸,
|
||||
// 否则闸的「恒放行」会把「回落默认」静默变成「无上限」(构造处已保证闸在归一之后建)。
|
||||
func TestNormalizeReviewConfigFillsT1ThrottleDefaults(t *testing.T) {
|
||||
t.Parallel()
|
||||
for _, cfg := range []ReviewConfig{
|
||||
{},
|
||||
{T1PerWindow: -1, T1Window: -time.Second},
|
||||
{T1PerWindow: 7},
|
||||
{T1Window: 3 * time.Minute},
|
||||
} {
|
||||
got := normalizeReviewConfig(cfg)
|
||||
if got.T1PerWindow <= 0 {
|
||||
t.Fatalf("normalize(%+v).T1PerWindow = %d,非正值直达闸会被读成「无上限」", cfg, got.T1PerWindow)
|
||||
}
|
||||
if got.T1Window <= 0 {
|
||||
t.Fatalf("normalize(%+v).T1Window = %s,同上", cfg, got.T1Window)
|
||||
}
|
||||
}
|
||||
if got := normalizeReviewConfig(ReviewConfig{}); got.T1PerWindow != defaultT1PerWindow ||
|
||||
got.T1Window != defaultT1Window {
|
||||
t.Fatalf("零值配置 = %d/%s, want %d/%s",
|
||||
got.T1PerWindow, got.T1Window, defaultT1PerWindow, defaultT1Window)
|
||||
}
|
||||
if got := normalizeReviewConfig(ReviewConfig{T1PerWindow: 7, T1Window: 3 * time.Minute}); got.T1PerWindow != 7 ||
|
||||
got.T1Window != 3*time.Minute {
|
||||
t.Fatalf("显式额度被改写:%d/%s", got.T1PerWindow, got.T1Window)
|
||||
}
|
||||
|
||||
// 构造后的生效值也必须已是默认额(而不是留 0 等运行时兜)。
|
||||
store := newTestStore(t)
|
||||
r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), ReviewConfig{}, nil, nil, nil)
|
||||
if r.cfg.T1PerWindow != defaultT1PerWindow || r.cfg.T1Window != defaultT1Window {
|
||||
t.Fatalf("NewPreInsertReviewer 的生效额度 = %d/%s, want 默认 %d/%s",
|
||||
r.cfg.T1PerWindow, r.cfg.T1Window, defaultT1PerWindow, defaultT1Window)
|
||||
}
|
||||
if allowed, first := r.t1Gate.allow(); !allowed || first {
|
||||
t.Fatal("默认额度下第一次就该放行")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user