fix(scheduler): 修复 OpenAI 高级调度器审计发现的正确性与性能问题

针对 #3692 合并后审计发现的问题集中修复:

- previous_response_id 剥离条件改为按 call_id 全覆盖校验,
  部分可重建的工具续链不再被误剥离(不受开关门控的行为回归)
- 粘性加权回退路径补分组归属校验并清理失效绑定,杜绝跨分组账号泄漏
- 账号列表页:无 OpenAI 账号时跳过分数计算、过滤池限定 openai 平台、
  负载批查合并为账号并集一次查询,消除全表扫描与 Redis N+1
- 订阅优先模式下常规池不可用时回退订阅池等待计划,
  busy-but-waitable 的订阅账号不再导致请求硬失败
- TopK/权重 DB 覆盖显式受总开关门控,与兄弟子开关语义一致
- 前端未分组 OpenAI 账号回退展示基础分,不再显示 "-"
- ListAllWithFilters 等能力正式进入 AccountRepository/AdminService 接口,
  移除匿名接口断言与静默降级;负载批查失败补 warn 日志
- SelectAccountWithSchedulerForCapability 增加显式 previousResponseCanMove
  参数,移除 "previous_response_can_move" 魔法字符串哨兵
- 设置写入路径补"基础权重不得全为零"聚合校验;
  运行时设置批量读取失败的降级路径覆盖全部键并留痕
This commit is contained in:
shaw
2026-07-06 11:43:16 +08:00
parent 759332aa92
commit 0fd2e9216d
22 changed files with 798 additions and 96 deletions
+112 -61
View File
@@ -196,14 +196,6 @@ type AccountSchedulerGroupScore struct {
const accountListGroupUngroupedQueryValue = "ungrouped"
type openAIAccountSchedulerScorePoolLister interface {
ListOpenAISchedulableAccountsForSchedulerScore(ctx context.Context, groupID *int64) ([]service.Account, error)
}
type accountSchedulerScoreFilterPoolLister interface {
ListAccountsForSchedulerScoreFilter(ctx context.Context, platform, accountType, status, search string, groupID int64, privacyMode string) ([]service.Account, error)
}
func (h *AccountHandler) buildAccountResponseWithRuntime(ctx context.Context, account *service.Account) AccountWithConcurrency {
item := AccountWithConcurrency{
Account: dto.AccountFromService(account),
@@ -250,33 +242,27 @@ func (h *AccountHandler) buildAccountResponseWithRuntime(ctx context.Context, ac
return item
}
func (h *AccountHandler) scoreOpenAIAccountSchedulerPool(ctx context.Context, accounts []service.Account) map[int64]AccountSchedulerScore {
// scoreOpenAIAccountSchedulerPool 对池内 OpenAI 账号计算调度分数快照。
// loadMap 为共享的账号负载数据(含池内全部账号即可,多余条目无害);传 nil 时自行批查。
func (h *AccountHandler) scoreOpenAIAccountSchedulerPool(ctx context.Context, accounts []service.Account, loadMap map[int64]*service.AccountLoadInfo) map[int64]AccountSchedulerScore {
if len(accounts) == 0 {
return nil
}
openAIAccounts := make([]*service.Account, 0, len(accounts))
loadReq := make([]service.AccountWithConcurrency, 0, len(accounts))
for i := range accounts {
account := &accounts[i]
if account.Platform != service.PlatformOpenAI {
continue
}
openAIAccounts = append(openAIAccounts, account)
loadReq = append(loadReq, service.AccountWithConcurrency{
ID: account.ID,
MaxConcurrency: account.EffectiveLoadFactor(),
})
}
if len(openAIAccounts) == 0 {
return nil
}
loadMap := map[int64]*service.AccountLoadInfo{}
if h.concurrencyService != nil {
if batchLoad, err := h.concurrencyService.GetAccountsLoadBatch(ctx, loadReq); err == nil && batchLoad != nil {
loadMap = batchLoad
}
if loadMap == nil {
loadMap = h.fetchOpenAIAccountLoadMap(ctx, openAIAccounts)
}
var scores map[int64]service.OpenAIAccountSchedulerScoreSnapshot
@@ -297,6 +283,36 @@ func (h *AccountHandler) scoreOpenAIAccountSchedulerPool(ctx context.Context, ac
return result
}
// fetchOpenAIAccountLoadMap 一次性批查给定 OpenAI 账号的负载数据;
// 失败时记录日志并返回空表(分数按零负载计算,属可接受降级)。
func (h *AccountHandler) fetchOpenAIAccountLoadMap(ctx context.Context, openAIAccounts []*service.Account) map[int64]*service.AccountLoadInfo {
loadMap := map[int64]*service.AccountLoadInfo{}
if h.concurrencyService == nil || len(openAIAccounts) == 0 {
return loadMap
}
seen := make(map[int64]struct{}, len(openAIAccounts))
loadReq := make([]service.AccountWithConcurrency, 0, len(openAIAccounts))
for _, account := range openAIAccounts {
if account == nil {
continue
}
if _, ok := seen[account.ID]; ok {
continue
}
seen[account.ID] = struct{}{}
loadReq = append(loadReq, service.AccountWithConcurrency{
ID: account.ID,
MaxConcurrency: account.EffectiveLoadFactor(),
})
}
if batchLoad, err := h.concurrencyService.GetAccountsLoadBatch(ctx, loadReq); err != nil {
slog.Warn("openai_scheduler_score_load_batch_failed", "error", err)
} else if batchLoad != nil {
loadMap = batchLoad
}
return loadMap
}
func (h *AccountHandler) buildOpenAIAccountSchedulerScores(
ctx context.Context,
accounts []service.Account,
@@ -309,12 +325,6 @@ func (h *AccountHandler) buildOpenAIAccountSchedulerScores(
filterPool = accounts
}
baseScores := make(map[int64]*AccountSchedulerScore)
for accountID, score := range h.scoreOpenAIAccountSchedulerPool(ctx, filterPool) {
copiedScore := score
baseScores[accountID] = &copiedScore
}
pageOpenAIAccountIDs := make(map[int64]struct{})
groupIDs := make(map[int64]struct{})
for i := range accounts {
@@ -338,7 +348,48 @@ func (h *AccountHandler) buildOpenAIAccountSchedulerScores(
}
}
if len(pageOpenAIAccountIDs) == 0 {
return baseScores, nil
return nil, nil
}
// 先取各分组池,再对"过滤池 ∪ 分组池"的账号并集做一次负载批查,
// 避免每个池各查一次 Redis 的 N+1。
groupIDList := make([]int64, 0, len(groupIDs))
for groupID := range groupIDs {
groupIDList = append(groupIDList, groupID)
}
sort.Slice(groupIDList, func(i, j int) bool { return groupIDList[i] < groupIDList[j] })
groupPools := make(map[int64][]service.Account, len(groupIDList))
if h.adminService != nil {
for _, groupID := range groupIDList {
gid := groupID
pool, err := h.adminService.ListOpenAISchedulableAccountsForSchedulerScore(ctx, &gid)
if err != nil {
slog.Warn("openai_scheduler_group_score_pool_failed", "group_id", gid, "error", err)
continue
}
groupPools[gid] = pool
}
}
loadUnion := make([]*service.Account, 0, len(filterPool))
collectOpenAIAccounts := func(pool []service.Account) {
for i := range pool {
if pool[i].Platform == service.PlatformOpenAI {
loadUnion = append(loadUnion, &pool[i])
}
}
}
collectOpenAIAccounts(filterPool)
for _, pool := range groupPools {
collectOpenAIAccounts(pool)
}
loadMap := h.fetchOpenAIAccountLoadMap(ctx, loadUnion)
baseScores := make(map[int64]*AccountSchedulerScore)
for accountID, score := range h.scoreOpenAIAccountSchedulerPool(ctx, filterPool, loadMap) {
copiedScore := score
baseScores[accountID] = &copiedScore
}
groupScoresByAccount := make(map[int64][]AccountSchedulerGroupScore)
@@ -346,7 +397,7 @@ func (h *AccountHandler) buildOpenAIAccountSchedulerScores(
if len(pool) == 0 {
return
}
scores := h.scoreOpenAIAccountSchedulerPool(ctx, pool)
scores := h.scoreOpenAIAccountSchedulerPool(ctx, pool, loadMap)
for accountID, schedulerScore := range scores {
if _, ok := pageOpenAIAccountIDs[accountID]; !ok {
continue
@@ -365,37 +416,27 @@ func (h *AccountHandler) buildOpenAIAccountSchedulerScores(
}
}
if lister, ok := h.adminService.(openAIAccountSchedulerScorePoolLister); ok {
groupIDList := make([]int64, 0, len(groupIDs))
for groupID := range groupIDs {
groupIDList = append(groupIDList, groupID)
for _, groupID := range groupIDList {
gid := groupID
pool, ok := groupPools[gid]
if !ok {
continue
}
sort.Slice(groupIDList, func(i, j int) bool { return groupIDList[i] < groupIDList[j] })
for _, groupID := range groupIDList {
gid := groupID
pool, err := lister.ListOpenAISchedulableAccountsForSchedulerScore(ctx, &gid)
if err != nil {
slog.Warn("openai_scheduler_group_score_pool_failed", "group_id", gid, "error", err)
continue
}
groupNameByID := make(map[int64]string)
groupPriorityByAccount := make(map[int64]int)
for i := range pool {
account := &pool[i]
for _, accountGroup := range account.AccountGroups {
if accountGroup.GroupID != gid {
continue
}
groupPriorityByAccount[account.ID] = accountGroup.Priority
if accountGroup.Group != nil {
groupNameByID[gid] = accountGroup.Group.Name
}
groupNameByID := make(map[int64]string)
groupPriorityByAccount := make(map[int64]int)
for i := range pool {
account := &pool[i]
for _, accountGroup := range account.AccountGroups {
if accountGroup.GroupID != gid {
continue
}
groupPriorityByAccount[account.ID] = accountGroup.Priority
if accountGroup.Group != nil {
groupNameByID[gid] = accountGroup.Group.Name
}
}
scoreGroupPool(&gid, groupNameByID, groupPriorityByAccount, pool)
}
scoreGroupPool(&gid, groupNameByID, groupPriorityByAccount, pool)
}
for accountID := range groupScoresByAccount {
@@ -417,11 +458,9 @@ func (h *AccountHandler) listAccountSchedulerScoreFilterPool(
if h.adminService == nil || (platform != "" && platform != service.PlatformOpenAI) {
return nil
}
lister, ok := h.adminService.(accountSchedulerScoreFilterPoolLister)
if !ok {
return nil
}
accounts, err := lister.ListAccountsForSchedulerScoreFilter(ctx, platform, accountType, status, search, groupID, privacyMode)
// 池只用于 OpenAI 分数计算(非 OpenAI 账号会在打分时被丢弃),
// 无论列表页平台过滤为何,查询一律限定 openai,避免无过滤时全表扫描。
accounts, err := h.adminService.ListAccountsForSchedulerScoreFilter(ctx, service.PlatformOpenAI, accountType, status, search, groupID, privacyMode)
if err != nil {
slog.Warn("openai_scheduler_filter_score_pool_failed", "error", err)
return nil
@@ -481,8 +520,20 @@ func (h *AccountHandler) List(c *gin.Context) {
var windowCosts map[int64]float64
var activeSessions map[int64]int
var rpmCounts map[int64]int
schedulerFilterPool := h.listAccountSchedulerScoreFilterPool(c.Request.Context(), platform, accountType, status, search, groupID, privacyMode)
schedulerScores, schedulerGroupScores := h.buildOpenAIAccountSchedulerScores(c.Request.Context(), accounts, schedulerFilterPool)
// 仅当前页存在 OpenAI 账号时才计算调度分数,避免为空结果付出池查询开销。
var schedulerScores map[int64]*AccountSchedulerScore
var schedulerGroupScores map[int64][]AccountSchedulerGroupScore
pageHasOpenAIAccounts := false
for i := range accounts {
if accounts[i].Platform == service.PlatformOpenAI {
pageHasOpenAIAccounts = true
break
}
}
if pageHasOpenAIAccounts {
schedulerFilterPool := h.listAccountSchedulerScoreFilterPool(c.Request.Context(), platform, accountType, status, search, groupID, privacyMode)
schedulerScores, schedulerGroupScores = h.buildOpenAIAccountSchedulerScores(c.Request.Context(), accounts, schedulerFilterPool)
}
// 始终获取并发数(Redis ZCARD,极低开销)
if h.concurrencyService != nil {
+1
View File
@@ -174,6 +174,7 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.
service.OpenAIUpstreamTransportHTTPSSE,
"",
false,
false,
service.PlatformGrok,
)
if err != nil {
@@ -145,6 +145,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
service.OpenAIUpstreamTransportAny,
service.OpenAIEndpointCapabilityChatCompletions,
false,
false,
requestPlatform,
)
if err != nil {
@@ -117,6 +117,7 @@ func (h *OpenAIGatewayHandler) Embeddings(c *gin.Context) {
service.OpenAIUpstreamTransportHTTPSSE,
service.OpenAIEndpointCapabilityEmbeddings,
false,
false,
)
if err != nil {
reqLog.Warn("openai_embeddings.account_select_failed",
@@ -110,6 +110,7 @@ func (h *OpenAIGatewayHandler) CountTokens(c *gin.Context) {
service.OpenAIUpstreamTransportAny,
service.OpenAIEndpointCapabilityChatCompletions,
false,
false,
openAICompatibleRequestPlatform(apiKey),
)
service.SetOpsLatencyMs(c, service.OpsAuthLatencyMsKey, time.Since(requestStart).Milliseconds())
@@ -350,6 +350,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
service.OpenAIUpstreamTransportAny,
service.OpenAIEndpointCapabilityChatCompletions,
requireCompact,
false,
requestPlatform,
)
if err != nil {
@@ -783,6 +784,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
service.OpenAIUpstreamTransportAny,
service.OpenAIEndpointCapabilityChatCompletions,
false,
false,
requestPlatform,
)
if err != nil {
@@ -1266,8 +1268,8 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
closeOpenAIClientWS(wsConn, coderws.StatusPolicyViolation, "previous_response_id must be a response.id (resp_*), not a message id")
return
}
firstMessageToolContext := service.ValidateFunctionCallOutputContextBytes(firstMessage)
previousResponseCanMove := !firstMessageToolContext.HasFunctionCallOutput || firstMessageToolContext.HasToolCallContext
firstMessageToolCoverage := service.AnalyzeToolCallOutputContextCoverageBytes(firstMessage)
previousResponseCanMove := !firstMessageToolCoverage.HasFunctionCallOutput || firstMessageToolCoverage.ContextCoversAllCallIDs
reqLog = reqLog.With(
zap.Bool("ws_ingress", true),
zap.String("model", reqModel),
@@ -1383,13 +1385,8 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
requiredTransport,
service.OpenAIEndpointCapabilityChatCompletions,
false,
previousResponseCanMove,
requestPlatform,
func() string {
if previousResponseCanMove {
return "previous_response_can_move"
}
return ""
}(),
)
if err != nil {
reqLog.Warn("openai.websocket_account_select_failed",