diff --git a/internal/lua/vm.go b/internal/lua/vm.go index 207b494..d9b2b7b 100644 --- a/internal/lua/vm.go +++ b/internal/lua/vm.go @@ -126,8 +126,11 @@ const ( // shrinkInterval is how often the VM janitor reclaims idle states. shrinkInterval = 30 * time.Second // shrinkGraceRounds is how many consecutive janitor rounds must see slack - // before anything is released, so a gap between requests is not mistaken - // for the end of a load period. + // before anything is released. It guards against tearing a pool down between + // two bursts — NOT against ordinary traffic: a pool serving one request at a + // time still has reclaimable slack, and requiring "no traffic at all" would + // pin a burst's leftover states forever on any busy gateway. The counter is + // therefore reset by GROWTH (real contention), not by a mere checkout. shrinkGraceRounds = 2 ) @@ -360,6 +363,9 @@ func (p *adapterPool) acquire() (*worker, error) { // sequential caller reusing one warm state. contended := p.created > 0 && p.inUse >= p.created p.created++ + // Growing is the signal that capacity is genuinely short, so any + // pending shrink decision is stale. + p.idleRounds = 0 extra := p.growPlanLocked(contended) p.mu.Unlock() if extra > 0 { @@ -391,8 +397,6 @@ func (p *adapterPool) checkoutLocked() { if p.inUse > p.peakInUse { p.peakInUse = p.inUse } - // Live demand invalidates any pending shrink decision. - p.idleRounds = 0 } func (p *adapterPool) release(w *worker) { diff --git a/internal/lua/vm_test.go b/internal/lua/vm_test.go index 1128844..7cbbcb1 100644 --- a/internal/lua/vm_test.go +++ b/internal/lua/vm_test.go @@ -947,9 +947,11 @@ func TestReclaimNeedsGraceRounds(t *testing.T) { } } -// TestCheckoutResetsGrace: live traffic between janitor rounds must cancel a -// pending shrink decision. -func TestCheckoutResetsGrace(t *testing.T) { +// TestGrowthResetsGrace: growth (real contention) must cancel a pending shrink +// decision, but ordinary sequential traffic must NOT — otherwise a busy gateway +// that always has a request in flight would pin every state a past burst +// created, which is exactly the leak this change removes. +func TestGrowthResetsGrace(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) @@ -969,11 +971,26 @@ func TestCheckoutResetsGrace(t *testing.T) { if n := vm.ReclaimIdleNow(); n != 0 { t.Fatalf("grace round reclaimed %d", n) } - // a request arrives: the grace counter resets + + // a lone sequential request reuses a warm state: no growth, so the pending + // shrink decision stands and the second round reclaims w, _ := p.acquire() p.release(w) + if n := vm.ReclaimIdleNow(); n == 0 { + t.Fatal("sequential traffic must not keep resetting the grace counter") + } + + // now rebuild slack and prove that GROWTH does reset it + ws = ws[:0] + for i := 0; i < 4; i++ { + w, _ := p.acquire() // concurrent holds force created to grow + ws = append(ws, w) + } + for _, w := range ws { + p.release(w) + } if n := vm.ReclaimIdleNow(); n != 0 { - t.Fatalf("traffic must reset the grace counter, reclaimed %d", n) + t.Fatalf("growth must reset the grace counter, reclaimed %d", n) } } @@ -1239,3 +1256,52 @@ func TestContentionBatchPrewarms(t *testing.T) { p.release(w1) p.release(w2) } + +// TestBurstLeftoverIsReclaimedUnderSteadyTraffic reproduces what production +// showed right after the first deploy: a burst grew the openai pool to 9 states, +// then steady low-concurrency traffic kept arriving and the pool never shrank, +// because any checkout reset the shrink grace counter. Slack must be reclaimed +// even while the gateway keeps serving requests one at a time. +func TestBurstLeftoverIsReclaimedUnderSteadyTraffic(t *testing.T) { + vm := NewVM(freshAdapterDir(t)) + if err := vm.Start(); err != nil { + t.Fatalf("start: %v", err) + } + defer vm.Stop() + vm.ConfigureConcurrency(map[string]int{"openai": 32}) + p := poolOf(t, vm, "openai") + + // burst: 9 concurrent holds + var ws []*worker + for i := 0; i < 9; i++ { + w, err := p.acquire() + if err != nil { + t.Fatalf("acquire: %v", err) + } + ws = append(ws, w) + } + for _, w := range ws { + p.release(w) + } + if created, _, _ := poolCounts(p); created < 9 { + t.Fatalf("burst should have grown the pool, created=%d", created) + } + + // steady state: one request between every janitor round, forever + for round := 0; round < 8; round++ { + w, err := p.acquire() + if err != nil { + t.Fatalf("round %d acquire: %v", round, err) + } + p.release(w) + vm.ReclaimIdleNow() + } + created, idle, inUse := poolCounts(p) + if inUse != 0 { + t.Fatalf("in_use = %d, want 0", inUse) + } + if created > residentWorkers+idleHeadroom { + t.Fatalf("burst leftovers were never reclaimed under steady traffic: created=%d idle=%d", + created, idle) + } +}