test(gateway): 修既有 flake — 根因在测试替身的取消契约,非产品 bug
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
This commit is contained in:
1 parent
b74de9fd86
commit
6dc1f218d5
1 file changed
+90
-5
@@ -26,11 +26,20 @@ type callRecord struct {
|
|||||||
// proceeds it records the call and returns a deterministic reply. A nil err
|
// 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
|
// produces a reply holding the *incoming* messages as text, so a test can assert
|
||||||
// exactly what context the resumed turn carried.
|
// 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 {
|
type gatedRunner struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
calls []callRecord
|
calls []callRecord
|
||||||
tickets chan struct{}
|
canceled int
|
||||||
err error
|
tickets chan struct{}
|
||||||
|
err error
|
||||||
}
|
}
|
||||||
|
|
||||||
func newGatedRunner(cap int) *gatedRunner {
|
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) {
|
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 {
|
select {
|
||||||
case <-g.tickets:
|
case <-g.tickets:
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
|
g.recordCanceled()
|
||||||
return nil, ctx.Err()
|
return nil, ctx.Err()
|
||||||
}
|
}
|
||||||
g.mu.Lock()
|
g.mu.Lock()
|
||||||
@@ -78,6 +92,47 @@ func (g *gatedRunner) count() int {
|
|||||||
return len(g.calls)
|
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.
|
// eventRec is one state transition captured by the recording sink.
|
||||||
type eventRec struct {
|
type eventRec struct {
|
||||||
sessionID, typ string
|
sessionID, typ string
|
||||||
@@ -556,7 +611,7 @@ func TestSubmitPendingBoundedRejectsOverflow(t *testing.T) {
|
|||||||
func TestInterruptCancelsTurnAndClearsPending(t *testing.T) {
|
func TestInterruptCancelsTurnAndClearsPending(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
run := newGatedRunner(2)
|
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 {
|
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr.Immediate {
|
||||||
t.Fatalf("A = %+v, want Immediate", sr)
|
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"))
|
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
|
// A new message after the interrupt must start fresh (Immediate), and the
|
||||||
// canceled chain must not have resumed B.
|
// canceled chain must not have resumed B.
|
||||||
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("C")}, loop.OwnerEnv()); !sr.Immediate {
|
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" {
|
if messageText(got[0].messages[0]) != "C" {
|
||||||
t.Fatalf("only surviving call = %v, want the C turn", messageTextList(got[0].messages))
|
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) {
|
func TestSubmitConcurrentSameSessionIsSafeAndBounded(t *testing.T) {
|
||||||
|
|||||||
Reference in new issue
Block a user