fix(seq): 修掉 seq_create 的 O(n²),并加极端压测

## 起因:1000×1000 压测直接跑爆

用户要求「1000 条序列 × 每条 1000 个组内 toolcall」。第一版跑满 8 分钟超时。
分阶段计时定位到瓶颈:

| 阶段 | 200 条 × 1000 工具 |
|---|---|
| 创建 | **27.0s**(135ms/条,**随序列数线性增长**) |
| 执行(组内 1000 并发) | 0.55s(20 万次调用,2.7µs/次) |
| 删除 | 4.7ms |

瓶颈在创建,不在执行。

## 根因

```go
// handlers.go:70 —— 每次 seq_create 之后
graphErr := p.store.CheckGraph()

// store.go:166 —— List() 全量 + 逐条 Load() 全部序列
```

1000 条各 250KB ⇒ 每次创建都重读 250MB 并反序列化。第 N 条的创建代价
随 N 线性增长,总计 O(n²)。

## ★ 走过的弯路:我一度建议「把校验挪到运行期」—— 那是错的

store.go:163 明确写着:

    两条检查(都必须在**建序列/保存**时做,而不是等运行):
      1. 每个 seq_call 的目标必须存在(不存在会在运行期才发现,浪费一整轮)

**校验时机是语义,不是性能旋钮。** 目标不存在若等到运行才发现,模型已经
白白花掉一整轮工具调用。性能问题不能靠挪语义来解。

## 修法:缓存调用边,Save 做 O(1) 增量

```go
// Store 新增
graph map[string][]string   // 序列名 → 它调用的目标(裸名)

// Save:  只更新这一条的边
s.graph[seq.Name] = edgesOf(seq)
// Delete: 移除这一条的边
delete(s.graph, name)
```

`callTargets` 只依赖 AST,不必每次从盘重建。**校验语义完全不变** —— 目标
存在性与三色 DFS 环检测都照旧在建序列时执行。

## 判据

- TestStoreGraphCacheKeepsSemantics  逐条钉住三个保证:目标存在性 ✓、
  环检测 ✓、删除后不再误报成环 ✓
  (这类优化最危险的失败模式是"校验还在跑但少查了某种情况")
- TestStoreSaveScalesLinearly       分段对比后半程/前半程每条耗时。
  ★ 判据自己改过一次:初版用「总耗时 ÷ 单条耗时」,而单条只有 48µs 时
    噪声占比过高,同一份代码两次跑出 84× 和 203× —— 判据不稳定时报的
    失败就是噪声,比没判据更糟。改成分段对比(平方时后半程会慢约 n/2
    倍,线性时基本持平),阈值 3 倍留足磁盘与 GC 抖动余量。
  实测 300 条:84~203× 单条(线性期望 300×),平方会是 90000×。

## 压测本身也修了两个自己的 bug

- 源文件目录与 store 目录分离时只改了写入侧,清理侧还指着 store 目录 ⇒
  报 "no such file"。看起来像文件被提前删了,真因是路径拼错。
- newE2EPlugin 的 runner 参数写死 *e2eRunner,压测换替身就编译不过 ⇒
  改为接受 seqRunner 接口。

## 压测规模

TestStressExtreme_ThousandSeqs 现为 1000 条 × 1000 toolcall(O(n²) 修复后
可跑)。判据全是**不变量**:每工具恰好调一次、1000 槽在交错延迟下仍按
声明序合并(并发下若按完成序合并必然错位)、删除后无残留。
This commit is contained in:
JianFeeeee
2026-09-27 15:58:54 +08:00
parent 698e25d10c
commit 428aad74db
4 changed files with 297 additions and 27 deletions

View File

@ -1,10 +1,13 @@
package seq
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
// 阶段 P3:序列的存储、调用图与跨序列调用。
@ -232,3 +235,149 @@ func TestStoreBlocksPathTraversal(t *testing.T) {
t.Error("store 目录外的文件被删掉了 —— 路径穿越已突破边界")
}
}
// ⑨ ★ 调用图缓存不得改变校验**语义**。
//
// 背景:CheckGraph 原来每次 s.List() + 逐条 s.Load(),把全部序列重新读盘
// 反序列化。实测 200 条×1000 工具时创建要 27s、平均 135ms/条且随序列数线性
// 增长(O(n²))。改成缓存调用边后 Save 只 O(1) 增量更新。
//
// ⚠️ 这类优化最危险的失败模式是"**语义悄悄变了**":校验还在跑,但少查了
// 某种情况。所以逐条钉住原本的三个保证。
func TestStoreGraphCacheKeepsSemantics(t *testing.T) {
// helper:造一条含 seq_call 边、指向 targets 的序列
mk := func(t *testing.T, st *Store, name string, targets ...string) {
t.Helper()
tools := make([]string, 0, len(targets))
// out 必须是**对象**(键→类型),不是字符串数组 ——
// 第一次写判据时用了 []string,被解析器正确拦下并给出可执行的报错。
outs := map[string]string{}
for i, tgt := range targets {
outs[fmt.Sprintf("o%d", i)] = "string"
tools = append(tools, fmt.Sprintf(
`{"tool": "seq_call", "args": {"target": %q, "group": "g"}, "as": "o%d"}`, tgt, i))
}
body := map[string]interface{}{
"name": name,
"groups": []interface{}{map[string]interface{}{
"name": "g",
"in": map[string]string{},
"out": outs,
"tools": strings.Join(tools, " ; ") + " ;",
}},
}
b, err := json.Marshal(body)
if err != nil {
t.Fatal(err)
}
seq, err := Parse(b)
if err != nil {
t.Fatalf("Parse(%s): %v", name, err)
}
if err := st.Save(seq); err != nil {
t.Fatalf("Save(%s): %v", name, err)
}
}
// ① 目标存在性仍生效
dir := t.TempDir()
st := NewStore(dir)
mk(t, st, "a", "#missing")
if err := st.CheckGraph(); err == nil {
t.Error("调用不存在的序列却通过了 CheckGraph —— 缓存漏了目标存在性检查")
} else if !strings.Contains(err.Error(), "missing") {
t.Errorf("错误信息应指出 missing,实际:%v", err)
}
// ② 环检测仍生效
dir2 := t.TempDir()
st2 := NewStore(dir2)
mk(t, st2, "a", "#b")
mk(t, st2, "b", "#a")
if err := st2.CheckGraph(); err == nil {
t.Error("a→b→a 成环却通过了 CheckGraph —— 缓存漏了环检测")
} else if !strings.Contains(err.Error(), "成环") {
t.Errorf("错误信息应指出成环,实际:%v", err)
}
// ③ 删除后缓存里的边同步移除,不再误报"成环"
if err := st2.Delete("b"); err != nil {
t.Fatal(err)
}
if err := st2.CheckGraph(); err != nil && strings.Contains(err.Error(), "成环") {
t.Errorf("b 已删除,不该再报成环:%v", err)
}
}
// ⑩ 保存必须 O(1) 增量:N 条序列的创建耗时应线性而非平方增长。
//
// 判据写成**比值**而非绝对耗时:绝对值随机器波动,比值只反映增长率。
func TestStoreSaveScalesLinearly(t *testing.T) {
if testing.Short() {
t.Skip("scaling test")
}
save := func(t *testing.T, st *Store, name string) {
t.Helper()
body := fmt.Sprintf(
`{"name":%q,"groups":[{"name":"g","in":{},"out":{"o0":"string"},`+
`"tools":"{\"tool\":\"cmd_run\",\"args\":{\"command\":\"ls\"},\"as\":\"o0\"} ;"}]}`,
name)
seq, err := Parse([]byte(body))
if err != nil {
t.Fatal(err)
}
if err := st.Save(seq); err != nil {
t.Fatal(err)
}
}
dir := t.TempDir()
st := NewStore(dir)
one := time.Now()
save(t, st, "probe")
single := time.Since(one)
dir2 := t.TempDir()
st2 := NewStore(dir2)
const n = 300
start := time.Now()
for i := 0; i < n; i++ {
save(t, st2, fmt.Sprintf("s%04d", i))
}
total := time.Since(start)
// 用**批内均分比**而不是"总/单条"。
//
// 为什么:单条只要几十微秒,n=1 那一次的耗时里进程噪声占比很高,
// 除出来的 ratio 抖动极大 —— 同一份代码两次跑出 84× 和 203×。
// 判据自己不稳定,报的失败就是噪声,比没有判据更糟。
// 改用"后半程每条耗时 vs 前半程每条耗时":平方增长会让后半程明显更慢,
// 线性增长则基本持平。
half := n / 2
_ = half
// 分段计时:重建一个 store,前半程和后半程各计一次
dir3 := t.TempDir()
st3 := NewStore(dir3)
var tFirst, tSecond time.Duration
start3 := time.Now()
for i := 0; i < half; i++ {
save(t, st3, fmt.Sprintf("f%04d", i))
}
tFirst = time.Since(start3)
mid := time.Now()
for i := 0; i < half; i++ {
save(t, st3, fmt.Sprintf("s%04d", i))
}
tSecond = time.Since(mid)
perFirst := float64(tFirst) / float64(half)
perSecond := float64(tSecond) / float64(half)
t.Logf("前半程 %v/条,后半程 %v/条(比值 %.2f;单条基准 %v,%d 条共 %v)",
time.Duration(int64(tFirst)/int64(half)), time.Duration(int64(tSecond)/int64(half)), perSecond/perFirst, single, n, total)
// 平方增长时后半程每条要贵约 n/2 倍;线性时基本持平。
// 阈值 3 倍给足余量(磁盘与 GC 抖动都在这个量级内)。
if perSecond > perFirst*3 {
t.Errorf("后半程每条 %v 是前半程 %v 的 %.1f 倍 —— 接近平方增长,调用图缓存没生效",
time.Duration(int64(tSecond)/int64(half)), time.Duration(int64(tFirst)/int64(half)), perSecond/perFirst)
}
}