fix(scheduler): keep account RPM config in the scheduler projection

`filterSchedulerExtra` 的白名单没有 `base_rpm` / `rpm_strategy` /
`rpm_sticky_buffer`,写进 `sched:meta:<id>` 的精简投影因此丢掉整套 RPM 配置。
候选过滤走 `ListSchedulableAccounts`,缓存命中时读的正是这份投影:
`isAccountSchedulableForRPM` 拿到的账号上 `GetBaseRPM()` 返回 0,`baseRPM <= 0`
直接放行,已配置限流的账号在选号阶段被当作未开启 RPM,单个自然分钟内被超额调度。
缓存未命中时回退数据库读到完整 extra,所以现象是间歇性的。

选中的账号随后由 `hydrateSelectedAccount` 从 `sched:acc:<id>` 重新水合,
RPM 计数与用户消息队列限速读的是完整账号,不受影响;缺的只是选号阶段这道门。

修法沿用投影里既有的处理方式(见 `openai_passthrough` 一组注释):把判定依赖的
键补进白名单,而不是在判定侧另开一条读取完整账号的路径。

回归测试覆盖序列化往返后的三个键:`tiered` 子用例证明缓冲被裁掉时红区会被误判成
黄区,`sticky_exempt` 子用例证明策略被裁掉时会退回三区判定。

Refs #7225
This commit is contained in:
Yoga Sakti
2026-09-18 13:37:58 +07:00
parent efe9aab1e4
commit 4805f069e4
2 changed files with 66 additions and 0 deletions
@@ -998,6 +998,11 @@ func filterSchedulerExtra(extra map[string]any) map[string]any {
"mixed_scheduling",
"window_cost_limit",
"window_cost_sticky_reserve",
// RPM 门与窗口费用门一样跑在本投影上:isAccountSchedulableForRPM 读 base_rpm,
// 缺失时 GetBaseRPM() 返回 0 并直接放行,已配置限流的账号会被超额调度。
"base_rpm",
"rpm_strategy",
"rpm_sticky_buffer",
"max_sessions",
"session_idle_timeout_minutes",
"openai_oauth_responses_websockets_v2_enabled",
@@ -1121,3 +1121,64 @@ func TestBuildSchedulerMetadataAccount_KeepsOpenAIPassthroughForModelGate(t *tes
})
}
}
// 调度投影必须保留账号的 RPM 配置。
//
// 候选过滤走 ListSchedulableAccounts,读的是 buildSchedulerMetadataAccount 产出的精简投影;
// isAccountSchedulableForRPM 靠 extra 上的 base_rpm 判定,缺失时 GetBaseRPM() 返回 0,
// baseRPM <= 0 直接放行,已配置限流的账号在选号阶段被当作未开启 RPM(#7225)。
// rpm_strategy / rpm_sticky_buffer 同理:只保留 base_rpm 会让三区判定退回默认值,
// 红区(不可调度)被误判成黄区(仅粘性),超额请求照样被转发。
func TestBuildSchedulerMetadataAccount_KeepsRPMFieldsForRPMGate(t *testing.T) {
newAccount := func(extra map[string]any) service.Account {
return service.Account{
ID: 7225,
Platform: service.PlatformAnthropic,
Type: service.AccountTypeOAuth,
Extra: extra,
}
}
roundTrip := func(t *testing.T, account service.Account) service.Account {
t.Helper()
// 走一遍真实的序列化/反序列化路径(写入 sched:meta 再由 decodeCachedAccount 读回)。
payload, err := json.Marshal(buildSchedulerMetadataAccount(account))
require.NoError(t, err)
var restored service.Account
require.NoError(t, json.Unmarshal(payload, &restored))
return restored
}
t.Run("tiered", func(t *testing.T) {
account := newAccount(map[string]any{
"base_rpm": 10,
"rpm_strategy": "tiered",
"rpm_sticky_buffer": 1,
})
require.Equal(t, service.WindowCostNotSchedulable, account.CheckRPMSchedulability(11),
"前置条件:base_rpm=10 + 缓冲 1,当前 11 次已进红区")
restored := roundTrip(t, account)
require.Equal(t, 10, restored.GetBaseRPM())
require.Equal(t, "tiered", restored.GetRPMStrategy())
require.Equal(t, 1, restored.GetRPMStickyBuffer())
require.Equal(t, service.WindowCostNotSchedulable, restored.CheckRPMSchedulability(11),
"投影裁掉 RPM 配置会让已配置限流的账号在候选过滤阶段被放行")
})
t.Run("sticky_exempt", func(t *testing.T) {
account := newAccount(map[string]any{
"base_rpm": 10,
"rpm_strategy": "sticky_exempt",
})
require.Equal(t, service.WindowCostStickyOnly, account.CheckRPMSchedulability(100),
"前置条件:粘性豁免策略没有红区")
restored := roundTrip(t, account)
require.Equal(t, 10, restored.GetBaseRPM())
require.Equal(t, "sticky_exempt", restored.GetRPMStrategy())
require.Equal(t, service.WindowCostStickyOnly, restored.CheckRPMSchedulability(100),
"投影裁掉 rpm_strategy 会让粘性豁免账号退回三区判定")
})
}