package scheduler import ( "context" "errors" "fmt" "strings" "sync/atomic" "testing" "time" "llmsproxy/internal/types" ) // fakeProvider is an in-memory Provider used to exercise chain scheduling // deterministically without a Lua runtime. type fakeProvider struct { name string model string pref atomic.Int64 available atomic.Bool busy atomic.Bool fail atomic.Bool chatHits atomic.Int64 // probeable makes an unavailable provider eligible as a half-cooldown // probe candidate. probePermit is the single-token permit; probeClaims // counts how many times it was handed out and probeDones how many times it // was returned, so tests can assert the permit is always released. probeable atomic.Bool probePermit atomic.Bool probeClaims atomic.Int64 probeDones atomic.Int64 } func fakeProv(name, model string) *fakeProvider { f := &fakeProvider{name: name, model: model} f.available.Store(true) return f } func (f *fakeProvider) Name() string { return f.name } func (f *fakeProvider) ModelFor(reqModel string) string { return f.model } func (f *fakeProvider) ModelAvailable(model string) bool { return f.available.Load() } func (f *fakeProvider) ModelSchedulable(model string) (bool, bool) { if f.available.Load() { return true, false } if !f.probeable.Load() { return false, false } if f.probePermit.CompareAndSwap(false, true) { f.probeClaims.Add(1) return true, true } return false, false } func (f *fakeProvider) ProbeDone(model string) { f.probeDones.Add(1) f.probePermit.Store(false) } func (f *fakeProvider) Pref(model string) int64 { return f.pref.Load() } func (f *fakeProvider) Chat(ctx context.Context, req *types.ChatRequest) (*types.UnifiedResponse, error) { f.chatHits.Add(1) if f.busy.Load() { return nil, types.ErrBusy } if f.fail.Load() { return nil, fmt.Errorf("upstream error") } return &types.UnifiedResponse{Content: f.name, FinishReason: "stop", TokenUsage: types.TokenUsage{Prompt: 1, Completion: 1, Total: 2}}, nil } func (f *fakeProvider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-chan types.UnifiedChunk, error) { if f.busy.Load() { return nil, types.ErrBusy } if f.fail.Load() { return nil, fmt.Errorf("upstream error") } ch := make(chan types.UnifiedChunk, 2) ch <- types.UnifiedChunk{Content: f.name} ch <- types.UnifiedChunk{Done: true} close(ch) return ch, nil } func (f *fakeProvider) Image(ctx context.Context, req *types.ImageGenRequest) (*types.UnifiedResponse, error) { return nil, errors.New("no image") } // TestDirectSkipsCooledCandidate: direct paths share the AUTO-chain rule that // cooldown is the only hard skip — a cooling candidate must never be hit, and // a fully cooling set must fail without touching upstream. func TestDirectSkipsCooledCandidate(t *testing.T) { hot := fakeProv("hot", "m") cold := fakeProv("cold", "m") cold.available.Store(false) s := New(2) req := &types.ChatRequest{ Model: "m", Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}}, } resp, src, _, err := s.Chat(context.Background(), []Provider{cold, hot}, req) if err != nil || src != "hot" { t.Fatalf("want hot to serve, got src=%q err=%v", src, err) } if resp == nil || resp.Content != "hot" { t.Fatalf("bad response: %#v", resp) } if cold.chatHits.Load() != 0 { t.Fatalf("cooled candidate must not be hit, got %d hits", cold.chatHits.Load()) } // all candidates cooled: direct path reports it instead of hitting upstream hot.available.Store(false) hits := hot.chatHits.Load() _, _, _, err = s.Chat(context.Background(), []Provider{cold, hot}, req) if err == nil || !strings.Contains(err.Error(), "cooling down") { t.Fatalf("want cooling-down error, got %v", err) } if hot.chatHits.Load() != hits || cold.chatHits.Load() != 0 { t.Fatalf("cooled candidates must not be hit, got hot=%d cold=%d", hot.chatHits.Load(), cold.chatHits.Load()) } } // lookup resolves (model, source) -> provider for chain builders in tests. type lookup func(m, s string) Provider func bySource(ps ...*fakeProvider) lookup { return func(m, s string) Provider { for _, p := range ps { if p.name == s { return p } } return nil } } func chatReq() *types.ChatRequest { return &types.ChatRequest{Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("hi")}}} } func TestBuildChainTierOrdering(t *testing.T) { a, b, c, d := fakeProv("s1", "a"), fakeProv("s2", "b"), fakeProv("s3", "c"), fakeProv("s4", "d") ch := BuildChain([]Rule{ {Tier: 1, Model: "a", Source: "s1"}, {Tier: 3, Model: "c", Source: "s3"}, {Tier: 2, Model: "b", Source: "s2"}, {Tier: 1, Model: "d", Source: "s4"}, {Tier: 9, Model: "gone", Source: "missing"}, }, bySource(a, b, c, d)) if len(ch.Tiers) != 3 { t.Fatalf("tiers = %d, want 3", len(ch.Tiers)) } if ch.Tiers[0].Tier != 1 || ch.Tiers[1].Tier != 2 || ch.Tiers[2].Tier != 3 { t.Fatalf("tier order = %d,%d,%d, want 1,2,3", ch.Tiers[0].Tier, ch.Tiers[1].Tier, ch.Tiers[2].Tier) } // same-tier slots keep configured order, unresolvable rule is dropped if len(ch.Tiers[0].Slots) != 2 || ch.Tiers[0].Slots[0].Model != "a" || ch.Tiers[0].Slots[1].Model != "d" { t.Fatalf("tier 1 slots = %+v", ch.Tiers[0].Slots) } } func TestChainRoundRobin(t *testing.T) { a, b := fakeProv("s1", "a"), fakeProv("s2", "b") ch := BuildChain([]Rule{ {Tier: 0, Model: "a", Source: "s1"}, {Tier: 0, Model: "b", Source: "s2"}, }, bySource(a, b)) s := New(0) var got []string for i := 0; i < 4; i++ { _, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil) if err != nil { t.Fatalf("iter %d: %v", i, err) } got = append(got, src) } want := []string{"s1", "s2", "s1", "s2"} for i := range want { if got[i] != want[i] { t.Fatalf("rr order = %v, want %v", got, want) } } } func TestChainPreferenceSinksButStaysReachable(t *testing.T) { neg := fakeProv("neg", "n") good := fakeProv("good", "g") neg.pref.Store(-5) good.fail.Store(true) ch := BuildChain([]Rule{ {Tier: 0, Model: "n", Source: "neg"}, {Tier: 0, Model: "g", Source: "good"}, }, bySource(neg, good)) s := New(0) resp, src, model, err := s.ChainChat(context.Background(), ch, chatReq(), nil) if err != nil { t.Fatalf("chain: %v", err) } // Round-robin starts at index 0 (neg). neg is available (pref=-5 > -20) // and succeeds. good is never tried because neg already won. if src != "neg" || model != "n" || resp.Content != "neg" { t.Fatalf("served src=%q model=%q content=%q", src, model, resp.Content) } if good.chatHits.Load() != 0 { t.Fatal("neg was tried first and succeeded; good must not be attempted") } // Second request: cursor advances. neg wins again (good hard-fails). resp2, src2, _, err2 := s.ChainChat(context.Background(), ch, chatReq(), nil) if err2 != nil { t.Fatalf("second chain: %v", err2) } if src2 != "neg" || resp2.Content != "neg" { t.Fatalf("second req src=%q content=%q", src2, resp2.Content) } } func TestChainBusySkipsWithoutPenalty(t *testing.T) { a, b := fakeProv("s1", "a"), fakeProv("s2", "b") a.busy.Store(true) ch := BuildChain([]Rule{ {Tier: 0, Model: "a", Source: "s1"}, {Tier: 0, Model: "b", Source: "s2"}, }, bySource(a, b)) s := New(0) _, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil) if err != nil { t.Fatalf("chain: %v", err) } if src != "s2" { t.Fatalf("src = %q, want s2", src) } if a.chatHits.Load() == 0 { t.Fatal("busy slot must have been attempted") } } func TestChainAllBusyBoundedWaitThenNextTier(t *testing.T) { oldWait, oldPoll := busyWait, busyPoll busyWait, busyPoll = 60*time.Millisecond, 10*time.Millisecond t.Cleanup(func() { busyWait, busyPoll = oldWait, oldPoll }) a, b, c := fakeProv("s1", "a"), fakeProv("s2", "b"), fakeProv("s3", "c") a.busy.Store(true) b.busy.Store(true) ch := BuildChain([]Rule{ {Tier: 1, Model: "a", Source: "s1"}, {Tier: 1, Model: "b", Source: "s2"}, {Tier: 2, Model: "c", Source: "s3"}, }, bySource(a, b, c)) s := New(0) t0 := time.Now() _, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil) el := time.Since(t0) if err != nil { t.Fatalf("chain: %v", err) } if src != "s3" { t.Fatalf("src = %q, want s3 (downgrade after bounded wait)", src) } if el > time.Second { t.Fatalf("busy wait not bounded: %v", el) } } func TestChainQuotaExhausted(t *testing.T) { a, b := fakeProv("s1", "a"), fakeProv("s2", "b") ch := BuildChain([]Rule{ {Tier: 0, Model: "a", Source: "s1", Quota: 100, Period: "hour"}, {Tier: 0, Model: "b", Source: "s2"}, }, bySource(a, b)) s := New(0) exhausted := func(sl *Slot) bool { return sl.Source == "s1" && sl.Quota > 0 } _, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), exhausted) if err != nil { t.Fatalf("chain: %v", err) } if src != "s2" { t.Fatalf("src = %q, want s2 (quota slot dropped)", src) } if a.chatHits.Load() != 0 { t.Fatal("quota-exhausted slot must not be called") } } func TestChainErrSummary(t *testing.T) { a, b, c := fakeProv("s1", "a"), fakeProv("s2", "b"), fakeProv("s3", "c") a.fail.Store(true) b.fail.Store(true) c.available.Store(false) // whole tier 1 cooling ch := BuildChain([]Rule{ {Tier: 0, Model: "a", Source: "s1"}, {Tier: 0, Model: "b", Source: "s2"}, {Tier: 1, Model: "c", Source: "s3"}, }, bySource(a, b, c)) s := New(0) _, _, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil) var ce *ChainErr if !errors.As(err, &ce) { t.Fatalf("err = %v, want *ChainErr", err) } if len(ce.Tiers) != 2 || ce.Tiers[0].Source != "s1" || ce.Tiers[1].Source != "s2" { t.Fatalf("tiers = %+v", ce.Tiers) } if len(ce.Skipped) != 1 { t.Fatalf("skipped = %+v", ce.Skipped) } msg := ce.Error() if !strings.Contains(msg, "s1") || !strings.Contains(msg, "s2") || !strings.Contains(msg, "tier 1: no schedulable slot") { t.Fatalf("summary = %q", msg) } } func TestChainStreamFallsBackBeforeFirstChunk(t *testing.T) { a, b := fakeProv("s1", "a"), fakeProv("s2", "b") a.fail.Store(true) ch := BuildChain([]Rule{ {Tier: 0, Model: "a", Source: "s1"}, {Tier: 0, Model: "b", Source: "s2"}, }, bySource(a, b)) s := New(0) chunks, src, model, err := s.ChainChatStream(context.Background(), ch, chatReq(), nil) if err != nil { t.Fatalf("chain stream: %v", err) } if src != "s2" || model != "b" { t.Fatalf("src=%q model=%q", src, model) } var text string for ck := range chunks { text += ck.Content } if text != "s2" { t.Fatalf("text = %q", text) } } func TestChainResetCooldownAfterSwap(t *testing.T) { a := fakeProv("s1", "a") ch := BuildChain([]Rule{{Tier: 0, Model: "a", Source: "s1"}}, bySource(a)) // a freshly built chain must schedule from index 0 (cursor starts at -1) if base := ch.Tiers[0].NextStart(); base != 0 { t.Fatalf("first start = %d, want 0", base) } } // ---- half-cooldown probe candidates (plan 阶段 1.3) ---- // TestChainProbeIsLastResort pins the ordering rule: a probe candidate must not // steal traffic from a healthy slot in the same tier. func TestChainProbeIsLastResort(t *testing.T) { healthy := fakeProv("healthy", "h") cooling := fakeProv("cooling", "c") cooling.available.Store(false) cooling.probeable.Store(true) ch := BuildChain([]Rule{ {Tier: 0, Model: "c", Source: "cooling"}, // configured FIRST on purpose {Tier: 0, Model: "h", Source: "healthy"}, }, bySource(healthy, cooling)) s := New(0) for i := 0; i < 3; i++ { _, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil) if err != nil { t.Fatalf("iter %d: %v", i, err) } if src != "healthy" { t.Fatalf("iter %d served by %q; a probe slot must never outrank a healthy one", i, src) } } if cooling.chatHits.Load() != 0 { t.Fatalf("probe slot was hit %d times while a healthy slot existed", cooling.chatHits.Load()) } if got := cooling.probeDones.Load(); got != cooling.probeClaims.Load() { t.Fatalf("probe permits leaked: claims=%d dones=%d", cooling.probeClaims.Load(), got) } } // TestChainProbeServesWhenNothingElseCan is the recovery path: with every normal // slot gone, the cooling slot's probe carries the request instead of the tier // failing outright. func TestChainProbeServesWhenNothingElseCan(t *testing.T) { cooling := fakeProv("cooling", "c") cooling.available.Store(false) cooling.probeable.Store(true) ch := BuildChain([]Rule{{Tier: 0, Model: "c", Source: "cooling"}}, bySource(cooling)) s := New(0) resp, src, model, err := s.ChainChat(context.Background(), ch, chatReq(), nil) if err != nil { t.Fatalf("probe must serve the request: %v", err) } if src != "cooling" || model != "c" { t.Fatalf("served by src=%q model=%q, want cooling/c", src, model) } if resp == nil || resp.Content != "cooling" { t.Fatalf("bad response: %#v", resp) } if cooling.chatHits.Load() != 1 { t.Fatalf("probe must issue exactly one request, got %d", cooling.chatHits.Load()) } if cooling.probeClaims.Load() != 1 || cooling.probeDones.Load() != 1 { t.Fatalf("permit accounting: claims=%d dones=%d, want 1/1", cooling.probeClaims.Load(), cooling.probeDones.Load()) } } // TestChainProbePermitReleasedOnFailure: a failed probe must still hand its // permit back, otherwise the slot could never be probed again. func TestChainProbePermitReleasedOnFailure(t *testing.T) { cooling := fakeProv("cooling", "c") cooling.available.Store(false) cooling.probeable.Store(true) cooling.fail.Store(true) ch := BuildChain([]Rule{{Tier: 0, Model: "c", Source: "cooling"}}, bySource(cooling)) s := New(0) if _, _, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil); err == nil { t.Fatal("expected the failing probe to surface an error") } if cooling.probeClaims.Load() != 1 || cooling.probeDones.Load() != 1 { t.Fatalf("permit accounting: claims=%d dones=%d, want 1/1", cooling.probeClaims.Load(), cooling.probeDones.Load()) } // permit is free again for the next attempt if _, _, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil); err == nil { t.Fatal("expected the second probe to fail too") } if cooling.probeClaims.Load() != 2 { t.Fatalf("permit must be reclaimable, claims=%d", cooling.probeClaims.Load()) } } // TestChainProbeDoesNotBlockTierFallthrough: a cooling tier-1 slot that is not // probeable must still let the request fall through to tier 2. func TestChainProbeDoesNotBlockTierFallthrough(t *testing.T) { cold := fakeProv("cold", "c") cold.available.Store(false) // cooling and NOT probeable backup := fakeProv("backup", "b") ch := BuildChain([]Rule{ {Tier: 1, Model: "c", Source: "cold"}, {Tier: 2, Model: "b", Source: "backup"}, }, bySource(cold, backup)) s := New(0) _, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil) if err != nil || src != "backup" { t.Fatalf("want fallthrough to backup, got src=%q err=%v", src, err) } if cold.chatHits.Load() != 0 { t.Fatal("non-probeable cooling slot must not be hit") } } // TestDirectProbeReleasesPermit covers the non-AUTO paths (Chat/ChatStream/Image). func TestDirectProbeReleasesPermit(t *testing.T) { cooling := fakeProv("cooling", "m") cooling.available.Store(false) cooling.probeable.Store(true) s := New(0) _, src, _, err := s.Chat(context.Background(), []Provider{cooling}, chatReq()) if err != nil || src != "cooling" { t.Fatalf("direct probe must serve: src=%q err=%v", src, err) } if cooling.probeClaims.Load() != 1 || cooling.probeDones.Load() != 1 { t.Fatalf("permit accounting: claims=%d dones=%d", cooling.probeClaims.Load(), cooling.probeDones.Load()) } } // TestChainProbeRoundRobinUnaffected: adding a probe tail must not disturb the // round-robin rotation over the healthy head. func TestChainProbeRoundRobinUnaffected(t *testing.T) { a, b := fakeProv("s1", "a"), fakeProv("s2", "b") cooling := fakeProv("s3", "c") cooling.available.Store(false) cooling.probeable.Store(true) ch := BuildChain([]Rule{ {Tier: 0, Model: "a", Source: "s1"}, {Tier: 0, Model: "b", Source: "s2"}, {Tier: 0, Model: "c", Source: "s3"}, }, bySource(a, b, cooling)) s := New(0) var got []string for i := 0; i < 4; i++ { _, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil) if err != nil { t.Fatalf("iter %d: %v", i, err) } got = append(got, src) } want := []string{"s1", "s2", "s1", "s2"} for i := range want { if got[i] != want[i] { t.Fatalf("rr order = %v, want %v (probe tail must not join the rotation)", got, want) } } }