From 6dc1f218d544d68f5dc02a59e0b3072a7cde7b17 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 20:30:38 +0800 Subject: [PATCH] =?UTF-8?q?test(gateway):=20=E4=BF=AE=E6=97=A2=E6=9C=89=20?= =?UTF-8?q?flake=20=E2=80=94=20=E6=A0=B9=E5=9B=A0=E5=9C=A8=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E6=9B=BF=E8=BA=AB=E7=9A=84=E5=8F=96=E6=B6=88=E5=A5=91?= =?UTF-8?q?=E7=BA=A6=EF=BC=8C=E9=9D=9E=E4=BA=A7=E5=93=81=20bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestInterruptCancelsTurnAndClearsPending 在纯 HEAD 与新代码上都约 7–15/50 失败（既有 flake，污染每次验收）。 - **根因（测试侧）**：测试替身 gatedRunner 阻塞在 select{tickets | ctx.Done()}；A 轮被 interrupt 取消时它已 park，ticket 释放后**同一调度步内两个 case 同时就绪**，Go 的 select 随机选 → 有时把**已被取消**的 turn 记成执行过（诊断重放 300 次：18 次失败，且每次被记执行的 ctx 都已 canceled） - **产品侧复核无 bug**：interruptLocked 在锁内 cancel + 清 pending + gen++；被取消/顶替的轮 completeTurn 命中 stale gen 直接返回，不碰状态、不续排 pending - **修法（仅测试）**：替身忠于真 runner 的取消契约（入口先查 ctx.Err()、两条取消路径都 recordCanceled、新增 waitCanceled）；测试在 Interrupt 后、释放 ticket 前**先等被取消轮确实停下**，断言不再依赖调度巧合；另加 waitIdle / run.count()==1 / session_idle 事件数==1（原有 4 处断言一字未删） - **稳定性**：修前 -count=50 四次 15/9/7/10 失败 → 修后 -count=50×8 + -count=100 + -count=300 **0/750**；-race 目标测试与整包均 ok - **变异验证**（防改断言变绿）：去掉 interrupt 的 cancel → 10/10 FAIL；不清 pending → FAIL；去掉 completeTurn 的 stale-gen 判断 → 10/10 FAIL --- .../gateway/session_orchestrator_test.go | 95 ++++++++++++++++++- 1 file changed, 90 insertions(+), 5 deletions(-) diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index ca1746f..3692645 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -26,11 +26,20 @@ type callRecord struct { // proceeds it records the call and returns a deterministic reply. A nil err // produces a reply holding the *incoming* messages as text, so a test can assert // exactly what context the resumed turn carried. +// +// Like the real loop runner, it never executes a turn whose context is already +// canceled, and it counts those canceled returns so a test can wait for an +// interrupted turn to actually stop (waitCanceled) before releasing the next +// ticket. Without that, a turn that was canceled *while parked in the select +// below* can still win the ticket case in the same scheduler step (both cases are +// ready) and be recorded as if it had run — a test-harness artifact, not an +// orchestrator ordering guarantee. type gatedRunner struct { - mu sync.Mutex - calls []callRecord - tickets chan struct{} - err error + mu sync.Mutex + calls []callRecord + canceled int + tickets chan struct{} + err error } func newGatedRunner(cap int) *gatedRunner { @@ -38,9 +47,14 @@ func newGatedRunner(cap int) *gatedRunner { } func (g *gatedRunner) Run(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error) { + if err := ctx.Err(); err != nil { + g.recordCanceled() + return nil, err + } select { case <-g.tickets: case <-ctx.Done(): + g.recordCanceled() return nil, ctx.Err() } g.mu.Lock() @@ -78,6 +92,47 @@ func (g *gatedRunner) count() int { return len(g.calls) } +// recordCanceled notes one invocation that returned because its context was +// canceled, i.e. a turn that never executed. +func (g *gatedRunner) recordCanceled() { + g.mu.Lock() + g.canceled++ + g.mu.Unlock() +} + +// waitCanceled blocks until at least n invocations have stopped because their +// context was canceled, reporting false when that never happens within the +// deadline. It is the barrier a test uses between an Interrupt and the release of +// the next ticket: once it holds, no canceled turn can still consume a ticket. +func (g *gatedRunner) waitCanceled(n int) bool { + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + g.mu.Lock() + ok := g.canceled >= n + g.mu.Unlock() + if ok { + return true + } + time.Sleep(time.Millisecond) + } + return false +} + +// waitIdle blocks until the session reports idle with an empty pending queue, +// reporting false when it never settles within the deadline. It lets a test assert +// the settled state deterministically instead of sleeping, and turns "a dropped +// turn was resumed anyway" into a test failure rather than a silent pass. +func waitIdle(o *Orchestrator, sessionID string) bool { + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if !o.Busy(sessionID) && o.QueueDepth(sessionID) == 0 { + return true + } + time.Sleep(time.Millisecond) + } + return false +} + // eventRec is one state transition captured by the recording sink. type eventRec struct { sessionID, typ string @@ -556,7 +611,7 @@ func TestSubmitPendingBoundedRejectsOverflow(t *testing.T) { func TestInterruptCancelsTurnAndClearsPending(t *testing.T) { t.Parallel() run := newGatedRunner(2) - o, _ := seedOrchestrator(run, 5) + o, sink := seedOrchestrator(run, 5) if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr.Immediate { t.Fatalf("A = %+v, want Immediate", sr) @@ -574,6 +629,14 @@ func TestInterruptCancelsTurnAndClearsPending(t *testing.T) { t.Fatalf("session not idle after interrupt: busy=%v depth=%d", o.Busy("s1"), o.QueueDepth("s1")) } + // The cancel must reach the in-flight turn (design/15 §3: interrupt is a + // graceful context cancel) and it must stop before any further turn is + // released: a canceled turn never executes, so it can never consume the ticket + // meant for the next one. + if !run.waitCanceled(1) { + t.Fatal("interrupted turn never observed its context cancellation") + } + // A new message after the interrupt must start fresh (Immediate), and the // canceled chain must not have resumed B. if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("C")}, loop.OwnerEnv()); !sr.Immediate { @@ -587,6 +650,28 @@ func TestInterruptCancelsTurnAndClearsPending(t *testing.T) { if messageText(got[0].messages[0]) != "C" { t.Fatalf("only surviving call = %v, want the C turn", messageTextList(got[0].messages)) } + + // C's chain settles with nothing left to run: had the interrupt failed to + // cancel A or to clear B, the surviving chain would have resumed a turn and the + // session would never go idle again. + if !waitIdle(o, "s1") { + t.Fatalf("session never settled idle: busy=%v depth=%d", o.Busy("s1"), o.QueueDepth("s1")) + } + if n := run.count(); n != 1 { + t.Fatalf("runner calls = %d, want 1 (neither the canceled A nor the dropped B may execute)", n) + } + // The canceled turn's completion is a no-op for session state (its generation is + // stale), so it must not emit the idle event of the turn now running: exactly + // one session_idle belongs to this scenario (C settling). + idles := 0 + for _, ty := range sink.types() { + if ty == EventSessionIdle { + idles++ + } + } + if idles != 1 { + t.Fatalf("session_idle events = %d, want 1 (a canceled turn must not settle the fresh chain)", idles) + } } func TestSubmitConcurrentSameSessionIsSafeAndBounded(t *testing.T) {