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 } 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) 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) } // higher pref (good) is tried first and hard-fails; the pass moves on to // the negative-pref slot, which sinks but stays reachable: with the real // provider its success would RecordSuccess (+1 pref, self-heal) 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("higher-pref slot must be tried first") } } 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) } }