mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-09-19 16:39:15 +00:00
fix(scheduler): rebalance the pref score so metered sources aren't written off silently
Report: an allowance-metered source (sensenova) had its deepseek model sitting at pref -10 while the WebUI showed zero failures. All three numbers were accurate, and they exposed three compounding problems. 1. Quota exhaustion cost the SAME score as a real failure. RecordQuotaExhausted intentionally avoids failCount (an exhausted allowance is not a fault), so the UI showed fails=0 / cooling=false — yet it deducted the full prefFailStep (5). For a metered source, running out of budget is an everyday event, so the score drifted deep negative with no visible cause. Now quota and 429 events cost prefQuotaStep (1): the cooldown already keeps the slot out of rotation until the window resets, the score only needs a mild preference for slots with budget. 2. Recovery was 5:1 asymmetric. A failure cost -5 but a success only +1, so a slot at -10 needed ten consecutive successes just to reach neutral — which it could never get, because a low score makes the scheduler not pick it in the first place (starvation). Success now rewards prefSuccessStep (2): recovery from -10 needs five successes, while a real failure still outweighs one. 3. No idle decay. A penalised slot kept its negative score forever once it stopped being selected. Pref() now applies lazy decay: after prefDecayAfter (2 min) of no outcome, the score drifts one step back toward 0 per interval (never past 0, never touches positive scores). Applied in Pref() and TryProbe(), so a naturally-recovered idle slot is schedulable again without needing a probe. The failure penalty itself is unchanged (prefFailStep=5), so genuinely broken upstreams are still marked as clearly worse than healthy ones. Tests: quota penalty lighter than failure; 5 quota resets stay well above the floor with failCount untouched; 429 is quota-class; recovery from -10 needs <=5; idle decay rehabilitates a written-off slot, stops at 0, and never drags a positive score; decay repeatedly lifts a slot off the prefMin floor. Updated the two pre-existing tests that asserted the old -5/129 +1 values.
This commit is contained in:
@ -34,9 +34,14 @@ import (
|
||||
// cooldown always expires and any success resets the state, so a fixed
|
||||
// upstream recovers on its own.
|
||||
type ModelState struct {
|
||||
pref atomic.Int64 // +1 per success / -5 per failure, clamped
|
||||
pref atomic.Int64 // +prefSuccessStep per success / -prefFailStep per failure, clamped
|
||||
failCount atomic.Int64
|
||||
cooldownUntil atomic.Int64 // unix seconds; 0 = schedulable
|
||||
// prefTouched is the unix second pref was last moved by an outcome. It is
|
||||
// the reference point for decay: a negative score with no traffic since
|
||||
// prefDecayAfter drifts one step toward 0 per elapsed interval, so a slot
|
||||
// cannot stay written off forever just because nothing selected it.
|
||||
prefTouched atomic.Int64
|
||||
// cooldownFrom is the unix second the current cooldown window opened. It
|
||||
// is the reference point for the half-cooldown probe gate: probing is
|
||||
// allowed only in the second half of [cooldownFrom, cooldownUntil), so a
|
||||
@ -50,10 +55,35 @@ type ModelState struct {
|
||||
}
|
||||
|
||||
const (
|
||||
// prefFailStep is the penalty for a REAL failure (5xx, transport error,
|
||||
// unusable adapter output): something is wrong with the upstream or the
|
||||
// model, so two of them are enough to push a slot to the back of its tier.
|
||||
prefFailStep = 5
|
||||
prefMin = -20
|
||||
prefMax = 20
|
||||
backoffBase = 5 * time.Second
|
||||
// prefQuotaStep is the penalty for running out of allowance (quota/TPM
|
||||
// exhausted). Deliberately much smaller than prefFailStep: an exhausted
|
||||
// quota means "no budget right now", not "this model is bad", and for
|
||||
// allowance-metered sources it is an everyday event rather than a fault.
|
||||
// Charging it the full failure penalty made such sources drift to a deeply
|
||||
// negative score with ZERO visible failures - RecordQuotaExhausted
|
||||
// intentionally does not touch failCount, so the WebUI showed fails=0 and
|
||||
// cooling=false while the score sat at -10.
|
||||
prefQuotaStep = 1
|
||||
// prefSuccessStep is the reward for a successful request. Recovery used to
|
||||
// be +1 against a -5 penalty, a 5:1 asymmetry that made a slot at -10 need
|
||||
// ten consecutive successes just to reach neutral. Rewarding +2 keeps
|
||||
// failures meaningful (a real failure still costs more than one success
|
||||
// earns) while letting a recovered slot return to normal rotation quickly.
|
||||
prefSuccessStep = 2
|
||||
prefMin = -20
|
||||
prefMax = 20
|
||||
// prefDecayAfter is how long a slot may sit at a negative score without any
|
||||
// traffic before the score drifts back toward neutral. Without this a slot
|
||||
// that was penalised and then simply not selected again (the score itself
|
||||
// makes it unattractive) stays penalised indefinitely: it has no way to earn
|
||||
// the successes that would rehabilitate it. Applied lazily on read, so there
|
||||
// are no timers.
|
||||
prefDecayAfter = 2 * time.Minute
|
||||
backoffBase = 5 * time.Second
|
||||
// backoffCap bounds the exponential schedule. It used to be 30min, which
|
||||
// in practice meant "a slot whose upstream blipped is unusable for half an
|
||||
// hour even though its quota recovered in seconds". Combined with the
|
||||
@ -142,8 +172,55 @@ func (s *ModelState) RecordFailure(auth bool) {
|
||||
if auth {
|
||||
pen *= 2
|
||||
}
|
||||
s.pref.Add(-pen)
|
||||
s.addPref(-pen)
|
||||
}
|
||||
|
||||
// addPref moves the preference score and stamps prefTouched, which restarts the
|
||||
// decay clock. Every outcome-driven score change must go through here so decay
|
||||
// never fires on a slot that is actively being scored.
|
||||
func (s *ModelState) addPref(delta int64) {
|
||||
s.pref.Add(delta)
|
||||
clampPref(&s.pref, prefMin, prefMax)
|
||||
s.prefTouched.Store(time.Now().Unix())
|
||||
}
|
||||
|
||||
// decayPref drifts a negative score back toward neutral once a slot has been
|
||||
// idle for prefDecayAfter. This closes a starvation loop: a penalised slot is
|
||||
// unattractive, so it stops being selected, so it never earns the successes that
|
||||
// would rehabilitate it, so it stays penalised. One step per elapsed interval,
|
||||
// never past 0, and positive scores are left alone (a proven-good slot should
|
||||
// not be dragged down for being idle).
|
||||
//
|
||||
// Called lazily from Pref(); no timers and no goroutines.
|
||||
func (s *ModelState) decayPref() {
|
||||
cur := s.pref.Load()
|
||||
if cur >= 0 {
|
||||
return
|
||||
}
|
||||
last := s.prefTouched.Load()
|
||||
if last <= 0 {
|
||||
// no reference point yet: start the clock instead of decaying blindly
|
||||
s.prefTouched.CompareAndSwap(0, time.Now().Unix())
|
||||
return
|
||||
}
|
||||
step := int64(prefDecayAfter / time.Second)
|
||||
if step <= 0 {
|
||||
return
|
||||
}
|
||||
elapsed := time.Now().Unix() - last
|
||||
if elapsed < step {
|
||||
return
|
||||
}
|
||||
steps := elapsed / step
|
||||
target := cur + steps
|
||||
if target > 0 {
|
||||
target = 0
|
||||
}
|
||||
// Advance prefTouched by exactly the consumed intervals so leftover time
|
||||
// still counts toward the next step.
|
||||
if s.pref.CompareAndSwap(cur, target) {
|
||||
s.prefTouched.Store(last + steps*step)
|
||||
}
|
||||
}
|
||||
|
||||
// openCooldown starts a fresh cooldown window of length cd. Recording both
|
||||
@ -164,8 +241,10 @@ func (s *ModelState) openCooldown(cd time.Duration) {
|
||||
func (s *ModelState) RecordRateLimit() {
|
||||
s.failCount.Add(1)
|
||||
s.openCooldown(rateLimitCooldown)
|
||||
s.pref.Add(-prefFailStep)
|
||||
clampPref(&s.pref, prefMin, prefMax)
|
||||
// A 429 is "too fast", not "broken": charge it like an allowance event, not
|
||||
// like a fault, or a throttled-but-healthy source sinks purely for being
|
||||
// popular.
|
||||
s.addPref(-prefQuotaStep)
|
||||
}
|
||||
|
||||
// RecordQuotaExhausted marks the model as out of allowance until its quota
|
||||
@ -185,20 +264,24 @@ func (s *ModelState) RecordQuotaExhausted(resetIn time.Duration) {
|
||||
// No failCount bump: an exhausted quota is not an error streak, and
|
||||
// letting it inflate failCount would make the NEXT real failure jump
|
||||
// straight to the capped backoff.
|
||||
s.pref.Add(-prefFailStep)
|
||||
clampPref(&s.pref, prefMin, prefMax)
|
||||
//
|
||||
// The score penalty is prefQuotaStep, not prefFailStep, for the same
|
||||
// reason: cooldown alone already keeps the slot out of rotation until the
|
||||
// window resets, so the score only needs to express a mild preference for
|
||||
// slots with budget left. Charging the full failure penalty here is what
|
||||
// drove allowance-metered sources to -10 and beyond with fails=0.
|
||||
s.addPref(-prefQuotaStep)
|
||||
}
|
||||
|
||||
// RecordSuccess resets the failure counter and cooldown and bumps the
|
||||
// preference score by one. A single success is enough to lift a slot off the
|
||||
// prefMin floor (prefMin+1 > prefMin), which is what lets one successful probe
|
||||
// return a written-off slot to normal rotation.
|
||||
// RecordSuccess resets the failure counter and cooldown and rewards the
|
||||
// preference score. A single success lifts a slot off the prefMin floor, which
|
||||
// is what lets one successful probe return a written-off slot to normal
|
||||
// rotation.
|
||||
func (s *ModelState) RecordSuccess() {
|
||||
s.failCount.Store(0)
|
||||
s.cooldownUntil.Store(0)
|
||||
s.cooldownFrom.Store(0)
|
||||
s.pref.Add(1)
|
||||
clampPref(&s.pref, prefMin, prefMax)
|
||||
s.addPref(prefSuccessStep)
|
||||
}
|
||||
|
||||
func (s *ModelState) reset() {
|
||||
@ -207,6 +290,7 @@ func (s *ModelState) reset() {
|
||||
s.cooldownFrom.Store(0)
|
||||
s.probeInFlight.Store(false)
|
||||
s.pref.Store(0)
|
||||
s.prefTouched.Store(0)
|
||||
}
|
||||
|
||||
// TryProbe attempts to claim the single probe permit for a model that normal
|
||||
@ -225,6 +309,7 @@ func (s *ModelState) TryProbe() bool {
|
||||
now := time.Now().Unix()
|
||||
until := s.cooldownUntil.Load()
|
||||
if until <= now {
|
||||
s.decayPref() // a naturally-recovered slot needs no probe
|
||||
if s.pref.Load() > prefMin {
|
||||
return false // schedulable normally, no probe needed
|
||||
}
|
||||
@ -251,8 +336,13 @@ func (s *ModelState) Probing() bool { return s.probeInFlight.Load() }
|
||||
// when the model is not cooling.
|
||||
func (s *ModelState) CooldownFrom() int64 { return s.cooldownFrom.Load() }
|
||||
|
||||
// Pref is the current preference score (higher = preferred).
|
||||
func (s *ModelState) Pref() int64 { return s.pref.Load() }
|
||||
// Pref is the current preference score (higher = preferred). Reading it also
|
||||
// applies idle decay, so a slot penalised long ago and never selected since
|
||||
// drifts back toward neutral instead of being written off permanently.
|
||||
func (s *ModelState) Pref() int64 {
|
||||
s.decayPref()
|
||||
return s.pref.Load()
|
||||
}
|
||||
|
||||
// FailCount is the number of consecutive failures.
|
||||
func (s *ModelState) FailCount() int64 { return s.failCount.Load() }
|
||||
|
||||
@ -242,8 +242,10 @@ func TestModelStateCooldownAndRecovery(t *testing.T) {
|
||||
if st.FailCount() != 0 {
|
||||
t.Fatalf("fail count after success = %d", st.FailCount())
|
||||
}
|
||||
if st.Pref() != 1-int64(prefFailStep) {
|
||||
t.Fatalf("pref after one failure (-5) then success (+1) = %d, want %d", st.Pref(), 1-int64(prefFailStep))
|
||||
// failure penalties prefFailStep, success rewards prefSuccessStep:
|
||||
// -5 then +2 = -3
|
||||
if st.Pref() != int64(prefSuccessStep)-int64(prefFailStep) {
|
||||
t.Fatalf("pref after one failure then success = %d, want %d", st.Pref(), int64(prefSuccessStep)-int64(prefFailStep))
|
||||
}
|
||||
}
|
||||
|
||||
@ -265,8 +267,8 @@ func TestModelStateRateLimitShortCooldown(t *testing.T) {
|
||||
if st.FailCount() != 15 {
|
||||
t.Fatalf("fail count = %d, want 15 (counted but not escalating)", st.FailCount())
|
||||
}
|
||||
if st.Pref() != -15*int64(prefFailStep) && st.Pref() > int64(prefMin) {
|
||||
t.Fatalf("pref = %d", st.Pref())
|
||||
if st.Pref() != -15*int64(prefQuotaStep) {
|
||||
t.Fatalf("pref after 15 x 429 = %d, want %d (a 429 is a quota-class event, not a fault)", st.Pref(), -15*int64(prefQuotaStep))
|
||||
}
|
||||
if p.ModelAvailable("m") {
|
||||
t.Fatal("model must be cooling right after a 429")
|
||||
@ -1044,3 +1046,140 @@ func TestRecoveryProbeBudgetIsBounded(t *testing.T) {
|
||||
t.Fatalf("3 probe windows must cost exactly 3 upstream requests, got %d", extra)
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaPenaltyIsLighterThanFailure is the regression for the report
|
||||
// "今天一次失败没看到,分数目前已经 -10 了" on an allowance-metered source
|
||||
// (sensenova). RecordQuotaExhausted deliberately does not touch failCount — the
|
||||
// WebUI therefore showed fails=0 and cooling=false — yet it used to charge the
|
||||
// SAME score penalty as a real failure, so two ordinary quota resets dragged the
|
||||
// slot to -10 with no visible fault. An exhausted allowance means "no budget
|
||||
// right now", not "this model is broken": the cooldown already keeps it out of
|
||||
// rotation, so the score penalty must be mild.
|
||||
func TestQuotaPenaltyIsLighterThanFailure(t *testing.T) {
|
||||
quota := &ModelState{}
|
||||
quota.RecordQuotaExhausted(time.Minute)
|
||||
fail := &ModelState{}
|
||||
fail.RecordFailure(false)
|
||||
|
||||
qp, fp := quota.Pref(), fail.Pref()
|
||||
if qp <= fp {
|
||||
t.Fatalf("a quota reset must cost less score than a real failure: quota=%d failure=%d", qp, fp)
|
||||
}
|
||||
if qp != -prefQuotaStep {
|
||||
t.Errorf("quota penalty = %d, want -%d", qp, prefQuotaStep)
|
||||
}
|
||||
if fp != -prefFailStep {
|
||||
t.Errorf("failure penalty = %d, want -%d", fp, prefFailStep)
|
||||
}
|
||||
|
||||
// The reported symptom: several quota resets in a day must NOT approach the
|
||||
// prefMin floor, since each one is a normal event for a metered source.
|
||||
s := &ModelState{}
|
||||
for i := 0; i < 5; i++ {
|
||||
s.RecordQuotaExhausted(time.Minute)
|
||||
}
|
||||
if got := s.Pref(); got <= prefMin/2 {
|
||||
t.Errorf("5 quota resets sank the score to %d (floor %d); a metered source would be written off", got, prefMin)
|
||||
}
|
||||
if s.FailCount() != 0 {
|
||||
t.Errorf("quota exhaustion must not inflate failCount, got %d", s.FailCount())
|
||||
}
|
||||
}
|
||||
|
||||
// TestRateLimitPenaltyIsMild covers the sibling case: a 429 means "too fast",
|
||||
// not "broken", so a popular-but-healthy slot must not sink for being throttled.
|
||||
func TestRateLimitPenaltyIsMild(t *testing.T) {
|
||||
s := &ModelState{}
|
||||
s.RecordRateLimit()
|
||||
if got := s.Pref(); got != -prefQuotaStep {
|
||||
t.Errorf("429 penalty = %d, want -%d (same class as a quota event)", got, prefQuotaStep)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSuccessRecoversFasterThanBefore pins the penalty/reward asymmetry. Reward
|
||||
// used to be +1 against a -5 penalty, so a slot at -10 needed TEN consecutive
|
||||
// successes just to reach neutral — in practice it never got there, because a
|
||||
// low score makes the slot unattractive to the scheduler in the first place.
|
||||
func TestSuccessRecoversFasterThanBefore(t *testing.T) {
|
||||
s := &ModelState{}
|
||||
s.RecordFailure(false)
|
||||
s.RecordFailure(false) // -10, the score from the bug report
|
||||
|
||||
n := 0
|
||||
for s.Pref() < 0 {
|
||||
s.RecordSuccess()
|
||||
n++
|
||||
if n > 50 {
|
||||
t.Fatal("score never recovered")
|
||||
}
|
||||
}
|
||||
// with +2 per success, -10 needs 5; the old +1 needed 10
|
||||
if n > 5 {
|
||||
t.Errorf("recovery from -10 took %d successes, want <= 5", n)
|
||||
}
|
||||
// a real failure must still outweigh a single success, or failures stop
|
||||
// meaning anything
|
||||
if prefSuccessStep >= prefFailStep {
|
||||
t.Errorf("success reward %d must stay below the failure penalty %d", prefSuccessStep, prefFailStep)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPrefDecayRehabilitatesIdleSlot closes the starvation loop: a penalised
|
||||
// slot is unattractive, so the scheduler stops picking it, so it never earns the
|
||||
// successes that would rehabilitate it. Without decay the score stays negative
|
||||
// forever even though the upstream may be perfectly healthy.
|
||||
func TestPrefDecayRehabilitatesIdleSlot(t *testing.T) {
|
||||
s := &ModelState{}
|
||||
s.RecordFailure(false)
|
||||
s.RecordFailure(false)
|
||||
if s.Pref() != -2*prefFailStep {
|
||||
t.Fatalf("setup: pref=%d", s.Pref())
|
||||
}
|
||||
|
||||
// backdate the last touch by three decay intervals
|
||||
step := int64(prefDecayAfter / time.Second)
|
||||
s.prefTouched.Store(time.Now().Unix() - 3*step)
|
||||
|
||||
got := s.Pref()
|
||||
if got != -2*prefFailStep+3 {
|
||||
t.Errorf("after 3 idle intervals pref=%d, want %d", got, -2*prefFailStep+3)
|
||||
}
|
||||
|
||||
// decay must never overshoot into positive territory
|
||||
s.prefTouched.Store(time.Now().Unix() - 1000*step)
|
||||
if got := s.Pref(); got != 0 {
|
||||
t.Errorf("decay overshot to %d, want to stop at 0", got)
|
||||
}
|
||||
|
||||
// a positive score must not be dragged down for being idle
|
||||
good := &ModelState{}
|
||||
good.RecordSuccess()
|
||||
good.RecordSuccess()
|
||||
before := good.Pref()
|
||||
good.prefTouched.Store(time.Now().Unix() - 1000*step)
|
||||
if after := good.Pref(); after != before {
|
||||
t.Errorf("idle decay must not touch a positive score: %d -> %d", before, after)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDecayLiftsSlotOffTheFloor verifies decay actually restores schedulability:
|
||||
// a slot at prefMin is refused by ModelAvailable, and before decay existed only
|
||||
// a probe could rescue it.
|
||||
func TestDecayLiftsSlotOffTheFloor(t *testing.T) {
|
||||
s := &ModelState{}
|
||||
for i := 0; i < 10; i++ {
|
||||
s.RecordFailure(false)
|
||||
}
|
||||
if s.Pref() != prefMin {
|
||||
t.Fatalf("setup: want floor %d, got %d", prefMin, s.Pref())
|
||||
}
|
||||
// clear the cooldown so only the score gates the slot
|
||||
s.cooldownUntil.Store(0)
|
||||
s.cooldownFrom.Store(0)
|
||||
|
||||
step := int64(prefDecayAfter / time.Second)
|
||||
s.prefTouched.Store(time.Now().Unix() - 2*step)
|
||||
if got := s.Pref(); got <= prefMin {
|
||||
t.Errorf("decay failed to lift the slot off the floor: %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user