test(flake): 4 条负载敏感用例确定性化（只改测试，生产代码零改动）
- #1 TestSpawnTimeoutFromToolDefault：墙钟竞态（Spawn→Wait 的调度抖动 vs 120ms 工具时限）⇒ 用「托底子代理占住唯一槽位 + gate 放行」把 running 变成由门控保证；并新增 DeadlineExceeded 断言。 - #2 TestParallelLimit：真实成因**不是**「同进程迭代状态泄漏」（任务书推测错了）—— 是 transitionRunning 与 runner 内 active++ 之间的**抢占窗口**（复现：-count=200 六次红、三进程并行 1/9）⇒ 改等 runner 内部 activeCount()>=2 这一**稳定态**，并新增「整轮排空峰值==2」。 - #3 TestBackthink_List_NewestFirst：真实成因**不是**时间戳并列（SQL 是 ORDER BY id DESC）—— 是**后台心跳 writer 在两次 List 之间插入新行**（决定性证据：心跳周期缩到 1ms ⇒ 21/200 红，且第二次读到的 id 更大）⇒ 不再 Start() 心跳，改同步 think 写 3 条。 - #4 TestClient_OnConnectErrorTriggersRedial：固定 60ms 睡眠 vs 每周期耗时不定的墙钟竞态（改动前基线 5/9 红）⇒ 改为等第 2 次拨号事件；顺带改正与内容不符的旧注释。 - 证据：逐条 -count=20 绿；整树连跑 3 次 59 ok/0 FAIL；**三进程并行整树 9/9 绿（改动前基线 5/9 红）**；GOMAXPROCS=1 ×4 进程加压；-count=200/500/500/300；-race 三包绿。 - 主会话独立复验：隔离 worktree build/vet 绿、四条 -count=20 全绿、两进程并行整树双 59 ok/0 FAIL；**变异独立复现**（TestParallelLimit 的 MaxParallel 2→3 ⇒ max concurrent runs = 3, want 2 红）。 - 顺带发现第 5 条既有 flake：internal/runtime/exec/TestTerminal_Timeout_Killed（改动前基线 2/9 红）；以及 audit/00-总览.md:321/372 对 #2/#3 成因的记载有误（DevDocs 由主会话更正）。
This commit is contained in:
AskaEth committed
1 parent
b19d798a2d
commit
b1f9b11d6e
4 files changed
+84
-19
No files matched your search
@@ -101,10 +101,21 @@ func TestClient_DialReadReconnect(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestClient_HeartbeatVerifiesPing sends an s=2-like heartbeat every 20ms and
|
||||
// asserts the dial URL is still the one provided.
|
||||
// TestClient_OnConnectErrorTriggersRedial 断言连接钩子失败会触发重拨：等到第二次
|
||||
// 拨号事件发生后才取消上下文，而不是先睡固定的 60ms 再看拨了几次——后者在并行
|
||||
// 负载下可能只够拨一次，等于把断言交给机器负载决定。
|
||||
func TestClient_OnConnectErrorTriggersRedial(t *testing.T) {
|
||||
const (
|
||||
// redialTarget 是断言「发生过重拨」所需的拨号次数：1 次首拨 + 1 次重拨。
|
||||
redialTarget = 2
|
||||
// dialEvents 是拨号事件通道的容量，保证 Run 在测试取消之前不会阻塞在投递上。
|
||||
dialEvents = 8
|
||||
)
|
||||
|
||||
var dials atomic.Int64
|
||||
// 每次拨号投递一个事件；测试等满 redialTarget 次再取消，因此断言的是
|
||||
// 「重拨确实发生了」这件事本身，与墙钟无关。
|
||||
dialed := make(chan struct{}, dialEvents)
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
conn, err := wsu.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
@@ -120,10 +131,13 @@ func TestClient_OnConnectErrorTriggersRedial(t *testing.T) {
|
||||
defer srv.Close()
|
||||
url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + "/testws"
|
||||
|
||||
// OnConnect always errors so the client keeps rejecting each session and the
|
||||
// dial fires repeatedly.
|
||||
// OnConnect 恒错，因此每次会话都被客户端拒绝、拨号反复发生。
|
||||
client := New(
|
||||
func(context.Context) (string, error) { dials.Add(1); return url, nil },
|
||||
func(context.Context) (string, error) {
|
||||
dials.Add(1)
|
||||
dialed <- struct{}{}
|
||||
return url, nil
|
||||
},
|
||||
Handler{
|
||||
OnConnect: func(ctx context.Context, c *Conn) error { return context.DeadlineExceeded },
|
||||
},
|
||||
@@ -131,17 +145,24 @@ func TestClient_OnConnectErrorTriggersRedial(t *testing.T) {
|
||||
WithLogger(logger.New(logger.Options{Out: io.Discard})),
|
||||
)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- client.Run(ctx) }()
|
||||
|
||||
time.Sleep(60 * time.Millisecond)
|
||||
for i := 0; i < redialTarget; i++ {
|
||||
select {
|
||||
case <-dialed:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatalf("timed out waiting for dial %d", i+1)
|
||||
}
|
||||
}
|
||||
cancel()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Run did not exit")
|
||||
}
|
||||
if dials.Load() < 2 {
|
||||
t.Errorf("dials = %d, want >= 2 (reconnect on failed connect hook)", dials.Load())
|
||||
if dials.Load() < redialTarget {
|
||||
t.Errorf("dials = %d, want >= %d (reconnect on failed connect hook)", dials.Load(), redialTarget)
|
||||
}
|
||||
}
|
||||
@@ -224,8 +224,11 @@ func TestParallelLimit(t *testing.T) {
|
||||
subs = append(subs, sub)
|
||||
}
|
||||
|
||||
// 等到两个并行槽位都被占用，再确认并发运行数从未超过 MaxParallel。
|
||||
waitFor(t, func() bool { return countStatus(subs, StatusRunning) == 2 }, waitTimeout)
|
||||
// 等到两个并行槽位都被占用，再确认并发运行数从未超过 MaxParallel。这里等的是
|
||||
// runner 内部的活跃计数，而不是「状态已翻成 running」：状态是在进入 runner 之前
|
||||
// 置位的，两者之间有一个调度窗口，用墙钟轮询去撞它必然偶发（仅看状态时读到过
|
||||
// maxActive = 0/1）。gate 未开，因此 active == 2 是一个稳定状态。
|
||||
waitFor(t, func() bool { return m.activeCount() >= 2 }, waitTimeout)
|
||||
if got := m.maxActiveCount(); got != 2 {
|
||||
t.Fatalf("max concurrent runs = %d, want 2 while gated", got)
|
||||
}
|
||||
@@ -246,6 +249,11 @@ func TestParallelLimit(t *testing.T) {
|
||||
if res.Total != 5 || res.Succeeded != 5 {
|
||||
t.Fatalf("merged = %d total / %d succeeded, want 5/5", res.Total, res.Succeeded)
|
||||
}
|
||||
// 峰值在整轮排空中都不得越界：这条覆盖放行后逐个进出槽位的阶段，
|
||||
// 因此它也校验每个运行体结束时确实归还了槽位。
|
||||
if got := m.maxActiveCount(); got != 2 {
|
||||
t.Fatalf("max concurrent runs = %d, want 2 across the whole drain", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPerAgentTimeout(t *testing.T) {
|
||||
|
||||
@@ -2,6 +2,7 @@ package orchestration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -172,12 +173,25 @@ func TestSpawnInjectsDataParagraph(t *testing.T) {
|
||||
|
||||
// TestSpawnTimeoutFromToolDefault 断言 WithSpawnTimeout（生产经
|
||||
// SUBAGENT_TIMEOUT_SEC 填充）覆盖调度器默认时限（02.4.4）。
|
||||
//
|
||||
// 「批次在 30ms 前台窗口内没有结算」这一条由门控保证：唯一的并行槽位先被一个
|
||||
// 托底子代理占住（它的时限是调度器默认的 1h，测试放行之前绝不结束），因此无论
|
||||
// 机器多忙，被派发的子代理在窗口内都不可能开始跑，更不可能结算。原先这条是靠
|
||||
// 「30ms 窗口 < 120ms 时限」的墙钟巧合成立的，负载一高就会先等到已结算的 "done"。
|
||||
func TestSpawnTimeoutFromToolDefault(t *testing.T) {
|
||||
t.Parallel()
|
||||
s := newScheduler(t, Options{MaxParallel: 1, PerAgentTimeout: time.Hour})
|
||||
s.runner = &mockRunner{gate: make(chan struct{})} // 阻塞直到 ctx 超时
|
||||
sp := NewSpawnTool(s, WithSpawnWait(30*time.Millisecond), WithSpawnTimeout(120*time.Millisecond))
|
||||
gate := make(chan struct{})
|
||||
s.runner = &mockRunner{gate: gate} // 阻塞直到测试放行或 ctx 超时
|
||||
|
||||
// 托底者占住唯一槽位；它在测试放行之前不会结束。
|
||||
holder, err := s.Spawn(context.Background(), AgentSpec{Goal: "hold the only slot"})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn holder: %v", err)
|
||||
}
|
||||
waitFor(t, func() bool { return statusOf(s, holder.ID) == StatusRunning }, waitTimeout)
|
||||
|
||||
sp := NewSpawnTool(s, WithSpawnWait(30*time.Millisecond), WithSpawnTimeout(120*time.Millisecond))
|
||||
raw, err := sp.Execute(context.Background(), map[string]any{"goal": "slow"})
|
||||
if err != nil {
|
||||
t.Fatalf("Execute: %v", err)
|
||||
@@ -187,11 +201,18 @@ func TestSpawnTimeoutFromToolDefault(t *testing.T) {
|
||||
t.Fatalf("Status = %q, want running", got.Status)
|
||||
}
|
||||
id := got.Agents[0].ID
|
||||
|
||||
// 放行托底者：槽位归还，被派发的子代理这才开始跑并在 120ms 上中止。
|
||||
// 该发送与托底者的接收配对（此时还没有别的接收者），因此它同样是确定性的。
|
||||
gate <- struct{}{}
|
||||
waitFor(t, func() bool { return statusOf(s, id) == StatusAborted }, waitTimeout)
|
||||
sub := s.pickOrFail(t, id)
|
||||
if !strings.Contains(sub.Summary, "120ms") {
|
||||
t.Fatalf("Summary = %q, want the 120ms tool-default deadline named", sub.Summary)
|
||||
}
|
||||
if !errors.Is(sub.Err, context.DeadlineExceeded) {
|
||||
t.Fatalf("Err = %v, want DeadlineExceeded", sub.Err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSpawnTimeoutPerCallOverride 断言按调用覆盖（timeout_seconds）优先于
|
||||
|
||||
@@ -266,18 +266,23 @@ func TestBackthink_SystemEventHighValueThinks(t *testing.T) {
|
||||
eventually(t, 2*time.Second, func() bool { return thinker.count() >= before+1 })
|
||||
}
|
||||
|
||||
// TestBackthink_List_NewestFirst 断言 List 严格由新到旧，且单行列表的头部与完整
|
||||
// 列表的头部是同一条记录。记录全部由测试同步写入，因此两次读取之间不存在并发
|
||||
// 写入者（见下方注释）。
|
||||
func TestBackthink_List_NewestFirst(t *testing.T) {
|
||||
t.Parallel()
|
||||
thinker := &mockThinker{text: "nothing to report"}
|
||||
b, _ := newBackthink(t, Config{Interval: 15 * time.Millisecond}, Deps{Thinker: thinker})
|
||||
b, _ := newBackthink(t, Config{}, Deps{Thinker: thinker})
|
||||
|
||||
b.Start(context.Background())
|
||||
defer func() { _ = b.Stop(context.Background()) }()
|
||||
// 三条可区分的记录，写入顺序即 sources 的顺序。
|
||||
sources := []string{"first", "second", "third"}
|
||||
// 记录由测试同步写入（直接调用 think），刻意不启动心跳：后台 ticker 会在
|
||||
// 两次 List 之间插入一条新记录，使「单行列表的头部 == 完整列表的头部」这个
|
||||
// 断言变成跟调度赛跑。此前偶红时读到的正是「刚插入的更新的 id」。
|
||||
for _, src := range sources {
|
||||
b.think(context.Background(), Trigger{Source: src, Summary: src, Import: ImportanceLow, Kind: TriggerHeartbeat})
|
||||
}
|
||||
|
||||
eventually(t, 2*time.Second, func() bool {
|
||||
recs, _ := b.List(context.Background(), 10)
|
||||
return len(recs) >= 2
|
||||
})
|
||||
top, err := b.List(context.Background(), 1)
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
@@ -294,11 +299,21 @@ func TestBackthink_List_NewestFirst(t *testing.T) {
|
||||
if top[0].ID != all[0].ID {
|
||||
t.Fatalf("List(1)[0].ID = %d, want %d (== newest)", top[0].ID, all[0].ID)
|
||||
}
|
||||
if len(all) != len(sources) {
|
||||
t.Fatalf("len(all) = %d, want %d (no writer ran between the two reads)", len(all), len(sources))
|
||||
}
|
||||
for i := 1; i < len(all); i++ {
|
||||
if all[i].ID >= all[i-1].ID {
|
||||
t.Fatalf("records not newest-first: %d before %d", all[i-1].ID, all[i].ID)
|
||||
}
|
||||
}
|
||||
// 最新在前不仅体现在 id 上，也体现在内容上：列表是写入顺序的逆序。
|
||||
for i := range sources {
|
||||
want := sources[len(sources)-1-i]
|
||||
if all[i].Trigger != want {
|
||||
t.Fatalf("all[%d].Trigger = %q, want %q (newest first)", i, all[i].Trigger, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackthink_Stop_NoLeak(t *testing.T) {
|
||||
|
||||
Reference in new issue
Block a user