gateway_scheduling.go 97 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685
  1. package service
  2. // 本文件由 gateway_service.go 纯移动拆分而来:账号选择与负载感知调度、窗口费用
  3. // 与 RPM 预取、候选排序/过滤、混合平台调度与选择失败诊断。仅做代码搬迁,
  4. // 无任何行为变更。
  5. import (
  6. "context"
  7. "fmt"
  8. "log/slog"
  9. mathrand "math/rand"
  10. "sort"
  11. "strings"
  12. "time"
  13. "github.com/Wei-Shaw/sub2api/internal/config"
  14. "github.com/Wei-Shaw/sub2api/internal/pkg/claude"
  15. "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
  16. "github.com/Wei-Shaw/sub2api/internal/pkg/logger"
  17. "github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
  18. )
  19. // SelectAccount 选择账号(粘性会话+优先级)
  20. func (s *GatewayService) SelectAccount(ctx context.Context, groupID *int64, sessionHash string) (*Account, error) {
  21. return s.SelectAccountForModel(ctx, groupID, sessionHash, "")
  22. }
  23. // SelectAccountForModel 选择支持指定模型的账号(粘性会话+优先级+模型映射)
  24. func (s *GatewayService) SelectAccountForModel(ctx context.Context, groupID *int64, sessionHash string, requestedModel string) (*Account, error) {
  25. return s.SelectAccountForModelWithExclusions(ctx, groupID, sessionHash, requestedModel, nil)
  26. }
  27. // SelectAccountForModelWithExclusions selects an account supporting the requested model while excluding specified accounts.
  28. func (s *GatewayService) SelectAccountForModelWithExclusions(ctx context.Context, groupID *int64, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}) (*Account, error) {
  29. // 优先检查 context 中的强制平台(/antigravity 路由)
  30. var platform string
  31. forcePlatform, hasForcePlatform := ctx.Value(ctxkey.ForcePlatform).(string)
  32. if hasForcePlatform && forcePlatform != "" {
  33. platform = forcePlatform
  34. } else if groupID != nil {
  35. group, resolvedGroupID, err := s.resolveGatewayGroup(ctx, groupID)
  36. if err != nil {
  37. return nil, err
  38. }
  39. if group == nil {
  40. return nil, ErrGroupNotFound
  41. }
  42. groupID = resolvedGroupID
  43. ctx = s.withGroupContext(ctx, group)
  44. platform = group.Platform
  45. if group.Platform == PlatformComposite {
  46. decision, ok, err := s.resolveCompositeRouteDecision(ctx, group, requestedModel, CompositeRouteEndpointAny)
  47. if err != nil {
  48. return nil, err
  49. }
  50. if !ok {
  51. return nil, fmt.Errorf("%w supporting model: %s (composite target platform unknown)", ErrNoAvailableAccounts, requestedModel)
  52. }
  53. platform = decision.TargetPlatform
  54. requestedModel = decision.UpstreamModel
  55. ctx = WithCompositeRouteDecision(ctx, decision)
  56. }
  57. } else {
  58. // 无分组时只使用原生 anthropic 平台
  59. platform = PlatformAnthropic
  60. }
  61. ctx = s.withGatewayProfitControlGate(ctx, groupID)
  62. // Claude Code 限制可能已将 groupID 解析为 fallback group,
  63. // 渠道限制预检查必须使用解析后的分组。
  64. if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) {
  65. slog.Warn("channel pricing restriction blocked request",
  66. "group_id", derefGroupID(groupID),
  67. "model", requestedModel)
  68. return nil, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
  69. }
  70. // anthropic/gemini 分组支持混合调度(包含启用了 mixed_scheduling 的 antigravity 账户)
  71. // 注意:强制平台模式不走混合调度
  72. if (platform == PlatformAnthropic || platform == PlatformGemini) && !hasForcePlatform {
  73. account, err := s.selectAccountWithMixedScheduling(ctx, groupID, sessionHash, requestedModel, excludedIDs, platform)
  74. if err != nil {
  75. return nil, err
  76. }
  77. return s.hydrateSelectedAccount(ctx, account)
  78. }
  79. // antigravity 分组、强制平台模式或无分组使用单平台选择
  80. // 注意:强制平台模式也必须遵守分组限制,不再回退到全平台查询
  81. account, err := s.selectAccountForModelWithPlatform(ctx, groupID, sessionHash, requestedModel, excludedIDs, platform)
  82. if err != nil {
  83. return nil, err
  84. }
  85. return s.hydrateSelectedAccount(ctx, account)
  86. }
  87. // SelectAccountWithLoadAwareness selects account with load-awareness and wait plan.
  88. // metadataUserID: 用于客户端亲和调度,从中提取客户端 ID
  89. // sub2apiUserID: 系统用户 ID,用于二维亲和调度
  90. func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, groupID *int64, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, metadataUserID string, sub2apiUserID int64) (*AccountSelectionResult, error) {
  91. // 调试日志:记录调度入口参数
  92. excludedIDsList := make([]int64, 0, len(excludedIDs))
  93. for id := range excludedIDs {
  94. excludedIDsList = append(excludedIDsList, id)
  95. }
  96. slog.Debug("account_scheduling_starting",
  97. "group_id", derefGroupID(groupID),
  98. "model", requestedModel,
  99. "session", shortSessionHash(sessionHash),
  100. "excluded_ids", excludedIDsList)
  101. cfg := s.schedulingConfig()
  102. // 检查 Claude Code 客户端限制(可能会替换 groupID 为降级分组)
  103. group, groupID, err := s.checkClaudeCodeRestriction(ctx, groupID)
  104. if err != nil {
  105. return nil, err
  106. }
  107. ctx = s.withGroupContext(ctx, group)
  108. ctx = s.withGatewayProfitControlGate(ctx, groupID)
  109. // Claude Code 限制可能已将 groupID 解析为 fallback group,
  110. // 渠道限制预检查必须使用解析后的分组。
  111. if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) {
  112. slog.Warn("channel pricing restriction blocked request",
  113. "group_id", derefGroupID(groupID),
  114. "model", requestedModel)
  115. return nil, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
  116. }
  117. var stickyAccountID int64
  118. var stickySource string
  119. if prefetch := prefetchedStickyAccountIDFromContext(ctx, groupID); prefetch > 0 {
  120. stickyAccountID = prefetch
  121. stickySource = "prefetch"
  122. } else if sessionHash != "" && s.cache != nil {
  123. if accountID, err := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), sessionHash); err == nil {
  124. stickyAccountID = accountID
  125. stickySource = "cache"
  126. }
  127. }
  128. // [DEBUG-STICKY] 调度器入口日志
  129. slog.Info("sticky.scheduler_entry",
  130. "group_id", derefGroupID(groupID),
  131. "session_hash", shortSessionHash(sessionHash),
  132. "sticky_account_id", stickyAccountID,
  133. "sticky_source", stickySource,
  134. "model", requestedModel,
  135. "load_batch", cfg.LoadBatchEnabled,
  136. "has_concurrency_svc", s.concurrencyService != nil,
  137. "excluded_count", len(excludedIDs),
  138. )
  139. if s.debugModelRoutingEnabled() && requestedModel != "" {
  140. groupPlatform := ""
  141. if group != nil {
  142. groupPlatform = group.Platform
  143. }
  144. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] select entry: group_id=%v group_platform=%s model=%s session=%s sticky_account=%d load_batch=%v concurrency=%v",
  145. derefGroupID(groupID), groupPlatform, requestedModel, shortSessionHash(sessionHash), stickyAccountID, cfg.LoadBatchEnabled, s.concurrencyService != nil)
  146. }
  147. if s.concurrencyService == nil || !cfg.LoadBatchEnabled {
  148. // 复制排除列表,用于会话限制拒绝时的重试
  149. localExcluded := make(map[int64]struct{})
  150. for k, v := range excludedIDs {
  151. localExcluded[k] = v
  152. }
  153. for {
  154. account, err := s.SelectAccountForModelWithExclusions(ctx, groupID, sessionHash, requestedModel, localExcluded)
  155. if err != nil {
  156. return nil, err
  157. }
  158. result, err := s.tryAcquireAccountSlot(ctx, account.ID, account.Concurrency)
  159. if err == nil && result.Acquired {
  160. // 获取槽位后检查会话限制(使用 sessionHash 作为会话标识符)
  161. if !s.checkAndRegisterSession(ctx, account, sessionHash) {
  162. result.ReleaseFunc() // 释放槽位
  163. localExcluded[account.ID] = struct{}{} // 排除此账号
  164. continue // 重新选择
  165. }
  166. return s.newSelectionResultFromHydratedAccount(ctx, account, true, result.ReleaseFunc, nil), nil
  167. }
  168. // 对于等待计划的情况,也需要先检查会话限制
  169. if !s.checkAndRegisterSession(ctx, account, sessionHash) {
  170. localExcluded[account.ID] = struct{}{}
  171. continue
  172. }
  173. if stickyAccountID > 0 && stickyAccountID == account.ID && s.concurrencyService != nil {
  174. waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, account.ID)
  175. if waitingCount < cfg.StickySessionMaxWaiting {
  176. return s.newSelectionResultFromHydratedAccount(ctx, account, false, nil, &AccountWaitPlan{
  177. AccountID: account.ID,
  178. MaxConcurrency: account.Concurrency,
  179. Timeout: cfg.StickySessionWaitTimeout,
  180. MaxWaiting: cfg.StickySessionMaxWaiting,
  181. }), nil
  182. }
  183. }
  184. return s.newSelectionResultFromHydratedAccount(ctx, account, false, nil, &AccountWaitPlan{
  185. AccountID: account.ID,
  186. MaxConcurrency: account.Concurrency,
  187. Timeout: cfg.FallbackWaitTimeout,
  188. MaxWaiting: cfg.FallbackMaxWaiting,
  189. }), nil
  190. }
  191. }
  192. platform, hasForcePlatform, err := s.resolvePlatform(ctx, groupID, group, requestedModel)
  193. if err != nil {
  194. return nil, err
  195. }
  196. preferOAuth := platform == PlatformGemini
  197. if s.debugModelRoutingEnabled() && requestedModel != "" && modelRoutingAppliesToTargetPlatform(platform) {
  198. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] load-aware enabled: group_id=%v model=%s session=%s platform=%s", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), platform)
  199. }
  200. accounts, useMixed, err := s.listSchedulableAccounts(ctx, groupID, platform, hasForcePlatform)
  201. if err != nil {
  202. return nil, err
  203. }
  204. if len(accounts) == 0 {
  205. return nil, ErrNoAvailableAccounts
  206. }
  207. ctx = s.withWindowCostPrefetch(ctx, accounts)
  208. ctx = s.withRPMPrefetch(ctx, accounts)
  209. // 提前构建 accountByID(供 Layer 1 和 Layer 1.5 使用)
  210. accountByID := make(map[int64]*Account, len(accounts))
  211. for i := range accounts {
  212. accountByID[accounts[i].ID] = &accounts[i]
  213. }
  214. isExcluded := func(accountID int64) bool {
  215. if excludedIDs == nil {
  216. return false
  217. }
  218. _, excluded := excludedIDs[accountID]
  219. return excluded
  220. }
  221. // upstream 计费基准的渠道模型限制以账号映射后的上游模型为准,只能逐账号判定;
  222. // 负载感知各层的候选过滤与粘性 gate 共用这一判定。
  223. needsUpstreamCheck := s.needsUpstreamChannelRestrictionCheck(ctx, groupID)
  224. isChannelRestricted := func(account *Account) bool {
  225. return needsUpstreamCheck && s.isUpstreamModelRestrictedByChannel(ctx, *groupID, account, requestedModel)
  226. }
  227. // 获取模型路由配置(anthropic / openai 目标平台;composite 分组按目标平台判断)
  228. var routingAccountIDs []int64
  229. if group != nil && requestedModel != "" &&
  230. modelRoutingAppliesToPlatform(platform, group.Platform) {
  231. routingAccountIDs = group.GetRoutingAccountIDs(requestedModel)
  232. if s.debugModelRoutingEnabled() {
  233. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] context group routing: group_id=%d model=%s enabled=%v rules=%d matched_ids=%v session=%s sticky_account=%d",
  234. group.ID, requestedModel, group.ModelRoutingEnabled, len(group.ModelRouting), routingAccountIDs, shortSessionHash(sessionHash), stickyAccountID)
  235. if len(routingAccountIDs) == 0 && group.ModelRoutingEnabled && len(group.ModelRouting) > 0 {
  236. keys := make([]string, 0, len(group.ModelRouting))
  237. for k := range group.ModelRouting {
  238. keys = append(keys, k)
  239. }
  240. sort.Strings(keys)
  241. const maxKeys = 20
  242. if len(keys) > maxKeys {
  243. keys = keys[:maxKeys]
  244. }
  245. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] context group routing miss: group_id=%d model=%s patterns(sample)=%v", group.ID, requestedModel, keys)
  246. }
  247. }
  248. }
  249. // ============ Layer 1: 模型路由优先选择(优先级高于粘性会话) ============
  250. if len(routingAccountIDs) > 0 && s.concurrencyService != nil {
  251. // 1. 过滤出路由列表中可调度的账号
  252. var routingCandidates []*Account
  253. var filteredExcluded, filteredMissing, filteredUnsched, filteredPlatform, filteredModelScope, filteredModelMapping, filteredChannelRestricted, filteredWindowCost int
  254. var modelScopeSkippedIDs []int64 // 记录因模型限流被跳过的账号 ID
  255. for _, routingAccountID := range routingAccountIDs {
  256. if isExcluded(routingAccountID) {
  257. filteredExcluded++
  258. continue
  259. }
  260. account, ok := accountByID[routingAccountID]
  261. if !ok || !s.isAccountSchedulableForSelection(account) {
  262. if !ok {
  263. filteredMissing++
  264. } else {
  265. filteredUnsched++
  266. }
  267. continue
  268. }
  269. if !s.isGatewayAccountProfitEligible(ctx, account) {
  270. continue
  271. }
  272. if !s.isAccountAllowedForPlatform(account, platform, useMixed) {
  273. filteredPlatform++
  274. continue
  275. }
  276. if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, account, requestedModel) {
  277. filteredModelMapping++
  278. continue
  279. }
  280. if isChannelRestricted(account) {
  281. filteredChannelRestricted++
  282. continue
  283. }
  284. if !s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) {
  285. filteredModelScope++
  286. modelScopeSkippedIDs = append(modelScopeSkippedIDs, account.ID)
  287. continue
  288. }
  289. // 配额检查
  290. if !s.isAccountSchedulableForQuota(account) {
  291. continue
  292. }
  293. // 窗口费用检查(非粘性会话路径)
  294. if !s.isAccountSchedulableForWindowCost(ctx, account, false) {
  295. filteredWindowCost++
  296. continue
  297. }
  298. // RPM 检查(非粘性会话路径)
  299. if !s.isAccountSchedulableForRPM(ctx, account, false) {
  300. continue
  301. }
  302. routingCandidates = append(routingCandidates, account)
  303. }
  304. if s.debugModelRoutingEnabled() {
  305. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] routed candidates: group_id=%v model=%s routed=%d candidates=%d filtered(excluded=%d missing=%d unsched=%d platform=%d model_scope=%d model_mapping=%d channel_restricted=%d window_cost=%d)",
  306. derefGroupID(groupID), requestedModel, len(routingAccountIDs), len(routingCandidates),
  307. filteredExcluded, filteredMissing, filteredUnsched, filteredPlatform, filteredModelScope, filteredModelMapping, filteredChannelRestricted, filteredWindowCost)
  308. if len(modelScopeSkippedIDs) > 0 {
  309. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] model_rate_limited accounts skipped: group_id=%v model=%s account_ids=%v",
  310. derefGroupID(groupID), requestedModel, modelScopeSkippedIDs)
  311. }
  312. }
  313. if len(routingCandidates) > 0 {
  314. // 1.5. 在路由账号范围内检查粘性会话
  315. if sessionHash != "" && stickyAccountID > 0 {
  316. slog.Debug("sticky.layer1_5_checking",
  317. "sticky_account_id", stickyAccountID,
  318. "in_routing_list", containsInt64(routingAccountIDs, stickyAccountID),
  319. "is_excluded", isExcluded(stickyAccountID),
  320. "in_account_map", func() bool { _, ok := accountByID[stickyAccountID]; return ok }(),
  321. "session", shortSessionHash(sessionHash),
  322. )
  323. if containsInt64(routingAccountIDs, stickyAccountID) && !isExcluded(stickyAccountID) {
  324. // 粘性账号在路由列表中,优先使用
  325. if stickyAccount, ok := accountByID[stickyAccountID]; ok {
  326. var stickyCacheMissReason string
  327. gatePass := s.isAccountSchedulableForSelection(stickyAccount) &&
  328. s.isGatewayAccountProfitEligible(ctx, stickyAccount) &&
  329. s.isAccountAllowedForPlatform(stickyAccount, platform, useMixed) &&
  330. (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, stickyAccount, requestedModel)) &&
  331. !isChannelRestricted(stickyAccount) &&
  332. s.isAccountSchedulableForModelSelection(ctx, stickyAccount, requestedModel) &&
  333. s.isAccountSchedulableForQuota(stickyAccount) &&
  334. s.isAccountSchedulableForWindowCost(ctx, stickyAccount, true)
  335. rpmPass := gatePass && s.isAccountSchedulableForRPM(ctx, stickyAccount, true)
  336. if rpmPass { // 粘性会话窗口费用+RPM 检查
  337. result, err := s.tryAcquireAccountSlot(ctx, stickyAccountID, stickyAccount.Concurrency)
  338. if err == nil && result.Acquired {
  339. // 会话数量限制检查
  340. if !s.checkAndRegisterSession(ctx, stickyAccount, sessionHash) {
  341. result.ReleaseFunc() // 释放槽位
  342. stickyCacheMissReason = "session_limit"
  343. // 继续到负载感知选择
  344. } else {
  345. slog.Debug("sticky.layer1_5_hit",
  346. "account_id", stickyAccountID,
  347. "session", shortSessionHash(sessionHash),
  348. "result", "slot_acquired",
  349. )
  350. if s.debugModelRoutingEnabled() {
  351. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] routed sticky hit: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), stickyAccountID)
  352. }
  353. return s.newSelectionResult(ctx, stickyAccount, true, result.ReleaseFunc, nil)
  354. }
  355. }
  356. if stickyCacheMissReason == "" {
  357. waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, stickyAccountID)
  358. if waitingCount < cfg.StickySessionMaxWaiting {
  359. // 会话数量限制检查(等待计划也需要占用会话配额)
  360. if !s.checkAndRegisterSession(ctx, stickyAccount, sessionHash) {
  361. stickyCacheMissReason = "session_limit"
  362. // 会话限制已满,继续到负载感知选择
  363. } else {
  364. // 必须走 newSelectionResult 以 hydrate 账号凭证:
  365. // 调度快照中的账号是精简版(OAuth token 等被剥离),
  366. // 直接返回会导致后续转发缺少凭证而鉴权失败。
  367. return s.newSelectionResult(ctx, stickyAccount, false, nil, &AccountWaitPlan{
  368. AccountID: stickyAccountID,
  369. MaxConcurrency: stickyAccount.Concurrency,
  370. Timeout: cfg.StickySessionWaitTimeout,
  371. MaxWaiting: cfg.StickySessionMaxWaiting,
  372. })
  373. }
  374. } else {
  375. stickyCacheMissReason = "wait_queue_full"
  376. }
  377. }
  378. // 粘性账号槽位满且等待队列已满,继续使用负载感知选择
  379. } else if !gatePass {
  380. stickyCacheMissReason = "gate_check"
  381. } else {
  382. stickyCacheMissReason = "rpm_red"
  383. }
  384. // 记录粘性缓存未命中的结构化日志
  385. if stickyCacheMissReason != "" {
  386. baseRPM := stickyAccount.GetBaseRPM()
  387. var currentRPM int
  388. if count, ok := rpmFromPrefetchContext(ctx, stickyAccount.ID); ok {
  389. currentRPM = count
  390. }
  391. logger.LegacyPrintf("service.gateway", "[StickyCacheMiss] reason=%s account_id=%d session=%s current_rpm=%d base_rpm=%d",
  392. stickyCacheMissReason, stickyAccountID, shortSessionHash(sessionHash), currentRPM, baseRPM)
  393. }
  394. } else {
  395. _ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  396. logger.LegacyPrintf("service.gateway", "[StickyCacheMiss] reason=account_cleared account_id=%d session=%s current_rpm=0 base_rpm=0",
  397. stickyAccountID, shortSessionHash(sessionHash))
  398. }
  399. }
  400. }
  401. // 2. 批量获取负载信息
  402. routingLoads := make([]AccountWithConcurrency, 0, len(routingCandidates))
  403. for _, acc := range routingCandidates {
  404. routingLoads = append(routingLoads, AccountWithConcurrency{
  405. ID: acc.ID,
  406. MaxConcurrency: acc.EffectiveLoadFactor(),
  407. })
  408. }
  409. routingLoadMap, _ := s.concurrencyService.GetAccountsLoadBatch(ctx, routingLoads)
  410. // 3. 按负载感知排序
  411. var routingAvailable []accountWithLoad
  412. for _, acc := range routingCandidates {
  413. loadInfo := routingLoadMap[acc.ID]
  414. if loadInfo == nil {
  415. loadInfo = &AccountLoadInfo{AccountID: acc.ID}
  416. }
  417. if loadInfo.LoadRate < 100 {
  418. routingAvailable = append(routingAvailable, accountWithLoad{account: acc, loadInfo: loadInfo})
  419. }
  420. }
  421. if len(routingAvailable) > 0 {
  422. // 排序:优先级 > 负载率 > 最后使用时间
  423. sort.SliceStable(routingAvailable, func(i, j int) bool {
  424. a, b := routingAvailable[i], routingAvailable[j]
  425. if a.account.Priority != b.account.Priority {
  426. return a.account.Priority < b.account.Priority
  427. }
  428. if a.loadInfo.LoadRate != b.loadInfo.LoadRate {
  429. return a.loadInfo.LoadRate < b.loadInfo.LoadRate
  430. }
  431. switch {
  432. case a.account.LastUsedAt == nil && b.account.LastUsedAt != nil:
  433. return true
  434. case a.account.LastUsedAt != nil && b.account.LastUsedAt == nil:
  435. return false
  436. case a.account.LastUsedAt == nil && b.account.LastUsedAt == nil:
  437. return false
  438. default:
  439. return a.account.LastUsedAt.Before(*b.account.LastUsedAt)
  440. }
  441. })
  442. shuffleWithinSortGroups(routingAvailable)
  443. // 4. 尝试获取槽位
  444. for _, item := range routingAvailable {
  445. result, err := s.tryAcquireAccountSlot(ctx, item.account.ID, item.account.Concurrency)
  446. if err == nil && result.Acquired {
  447. // 会话数量限制检查
  448. if !s.checkAndRegisterSession(ctx, item.account, sessionHash) {
  449. result.ReleaseFunc() // 释放槽位,继续尝试下一个账号
  450. continue
  451. }
  452. if sessionHash != "" && s.cache != nil {
  453. _ = s.bindGatewayStickySessionDuringSelection(ctx, groupID, sessionHash, item.account.ID)
  454. }
  455. if s.debugModelRoutingEnabled() {
  456. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] routed select: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), item.account.ID)
  457. }
  458. return s.newSelectionResult(ctx, item.account, true, result.ReleaseFunc, nil)
  459. }
  460. }
  461. // 5. 所有路由账号槽位满,尝试返回等待计划(选择负载最低的)
  462. // 遍历找到第一个满足会话限制的账号
  463. for _, item := range routingAvailable {
  464. if !s.checkAndRegisterSession(ctx, item.account, sessionHash) {
  465. continue // 会话限制已满,尝试下一个
  466. }
  467. if s.debugModelRoutingEnabled() {
  468. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] routed wait: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), item.account.ID)
  469. }
  470. return s.newSelectionResult(ctx, item.account, false, nil, &AccountWaitPlan{
  471. AccountID: item.account.ID,
  472. MaxConcurrency: item.account.Concurrency,
  473. Timeout: cfg.StickySessionWaitTimeout,
  474. MaxWaiting: cfg.StickySessionMaxWaiting,
  475. })
  476. }
  477. // 所有路由账号会话限制都已满,继续到 Layer 2 回退
  478. }
  479. // 路由列表中的账号都不可用(负载率 >= 100),继续到 Layer 2 回退
  480. logger.LegacyPrintf("service.gateway", "[ModelRouting] All routed accounts unavailable for model=%s, falling back to normal selection", requestedModel)
  481. }
  482. }
  483. // ============ Layer 1.5: 粘性会话(仅在无模型路由配置时生效) ============
  484. if len(routingAccountIDs) == 0 && sessionHash != "" && stickyAccountID > 0 && !isExcluded(stickyAccountID) {
  485. accountID := stickyAccountID
  486. if accountID > 0 && !isExcluded(accountID) {
  487. account, ok := accountByID[accountID]
  488. if ok {
  489. // 检查账户是否需要清理粘性会话绑定
  490. clearSticky := shouldClearStickySession(account, requestedModel)
  491. if clearSticky {
  492. slog.Debug("sticky.layer1_5_no_routing_clear",
  493. "account_id", accountID,
  494. "reason", "should_clear_sticky_session",
  495. "session", shortSessionHash(sessionHash),
  496. )
  497. _ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  498. }
  499. // 注意:不再检查 isAccountInGroup,因为 accountByID 已经从按分组过滤的
  500. // accounts 列表构建,账号一定在分组内。而 scheduler snapshot 缓存
  501. // 反序列化后 AccountGroups 字段为空,导致 isAccountInGroup 永远返回 false。
  502. platformOK := s.isAccountAllowedForPlatform(account, platform, useMixed)
  503. profitOK := s.isGatewayAccountProfitEligible(ctx, account)
  504. modelSupported := requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)
  505. channelOK := !isChannelRestricted(account)
  506. modelSchedulable := s.isAccountSchedulableForModelSelection(ctx, account, requestedModel)
  507. quotaOK := s.isAccountSchedulableForQuota(account)
  508. windowCostOK := s.isAccountSchedulableForWindowCost(ctx, account, true)
  509. rpmOK := s.isAccountSchedulableForRPM(ctx, account, true)
  510. schedulable := s.isAccountSchedulableForSelection(account)
  511. slog.Debug("sticky.layer1_5_no_routing_checks",
  512. "account_id", accountID,
  513. "session", shortSessionHash(sessionHash),
  514. "clear_sticky", clearSticky,
  515. "schedulable", schedulable,
  516. "platform_ok", platformOK,
  517. "profit_ok", profitOK,
  518. "model_supported", modelSupported,
  519. "channel_ok", channelOK,
  520. "model_schedulable", modelSchedulable,
  521. "quota_ok", quotaOK,
  522. "window_cost_ok", windowCostOK,
  523. "rpm_ok", rpmOK,
  524. )
  525. if !clearSticky && platformOK && profitOK && modelSupported && channelOK && modelSchedulable && quotaOK && windowCostOK && rpmOK && schedulable {
  526. result, err := s.tryAcquireAccountSlot(ctx, accountID, account.Concurrency)
  527. if err == nil && result.Acquired {
  528. // 会话数量限制检查
  529. if !s.checkAndRegisterSession(ctx, account, sessionHash) {
  530. result.ReleaseFunc() // 释放槽位,继续到 Layer 2
  531. slog.Debug("sticky.layer1_5_no_routing_miss",
  532. "account_id", accountID,
  533. "reason", "session_limit",
  534. "session", shortSessionHash(sessionHash),
  535. )
  536. } else {
  537. slog.Debug("sticky.layer1_5_no_routing_hit",
  538. "account_id", accountID,
  539. "session", shortSessionHash(sessionHash),
  540. "result", "slot_acquired",
  541. )
  542. if s.cache != nil {
  543. _ = s.cache.RefreshSessionTTL(ctx, derefGroupID(groupID), sessionHash, stickySessionTTL)
  544. }
  545. return s.newSelectionResult(ctx, account, true, result.ReleaseFunc, nil)
  546. }
  547. } else {
  548. slog.Debug("sticky.layer1_5_no_routing_slot_busy",
  549. "account_id", accountID,
  550. "session", shortSessionHash(sessionHash),
  551. )
  552. }
  553. waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, accountID)
  554. if waitingCount < cfg.StickySessionMaxWaiting {
  555. // 会话数量限制检查(等待计划也需要占用会话配额)
  556. if !s.checkAndRegisterSession(ctx, account, sessionHash) {
  557. // 会话限制已满,继续到 Layer 2
  558. } else {
  559. slog.Debug("sticky.layer1_5_no_routing_hit",
  560. "account_id", accountID,
  561. "session", shortSessionHash(sessionHash),
  562. "result", "wait_plan",
  563. )
  564. return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
  565. AccountID: accountID,
  566. MaxConcurrency: account.Concurrency,
  567. Timeout: cfg.StickySessionWaitTimeout,
  568. MaxWaiting: cfg.StickySessionMaxWaiting,
  569. })
  570. }
  571. }
  572. } else if !clearSticky {
  573. slog.Debug("sticky.layer1_5_no_routing_miss",
  574. "account_id", accountID,
  575. "reason", "gate_check_failed",
  576. "session", shortSessionHash(sessionHash),
  577. )
  578. }
  579. } else {
  580. slog.Debug("sticky.layer1_5_no_routing_miss",
  581. "account_id", accountID,
  582. "reason", "account_not_in_map",
  583. "session", shortSessionHash(sessionHash),
  584. )
  585. }
  586. }
  587. } else if len(routingAccountIDs) == 0 && sessionHash != "" {
  588. slog.Debug("sticky.layer1_5_no_routing_skip",
  589. "sticky_account_id", stickyAccountID,
  590. "is_excluded", func() bool { return stickyAccountID > 0 && isExcluded(stickyAccountID) }(),
  591. "session", shortSessionHash(sessionHash),
  592. "reason", func() string {
  593. if stickyAccountID == 0 {
  594. return "no_sticky_binding"
  595. }
  596. return "sticky_account_excluded"
  597. }(),
  598. )
  599. }
  600. // ============ Layer 2: 负载感知选择 ============
  601. slog.Debug("sticky.layer2_fallback",
  602. "session", shortSessionHash(sessionHash),
  603. "sticky_account_id", stickyAccountID,
  604. "reason", "sticky_not_used_falling_back_to_load_balance",
  605. "total_accounts", len(accounts),
  606. )
  607. candidates := make([]*Account, 0, len(accounts))
  608. channelRestrictedCount := 0
  609. for i := range accounts {
  610. acc := &accounts[i]
  611. if isExcluded(acc.ID) {
  612. continue
  613. }
  614. // Scheduler snapshots can be temporarily stale (bucket rebuild is throttled);
  615. // re-check schedulability here so recently rate-limited/overloaded accounts
  616. // are not selected again before the bucket is rebuilt.
  617. if !s.isAccountSchedulableForSelection(acc) {
  618. continue
  619. }
  620. if !s.isGatewayAccountProfitEligible(ctx, acc) {
  621. continue
  622. }
  623. if !s.isAccountAllowedForPlatform(acc, platform, useMixed) {
  624. continue
  625. }
  626. if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, acc, requestedModel) {
  627. continue
  628. }
  629. if isChannelRestricted(acc) {
  630. channelRestrictedCount++
  631. continue
  632. }
  633. if !s.isAccountSchedulableForModelSelection(ctx, acc, requestedModel) {
  634. continue
  635. }
  636. // 配额检查
  637. if !s.isAccountSchedulableForQuota(acc) {
  638. continue
  639. }
  640. // 窗口费用检查(非粘性会话路径)
  641. if !s.isAccountSchedulableForWindowCost(ctx, acc, false) {
  642. continue
  643. }
  644. // RPM 检查(非粘性会话路径)
  645. if !s.isAccountSchedulableForRPM(ctx, acc, false) {
  646. continue
  647. }
  648. candidates = append(candidates, acc)
  649. }
  650. if len(candidates) == 0 {
  651. if channelRestrictedCount > 0 {
  652. slog.Warn("channel pricing restriction blocked request",
  653. "group_id", derefGroupID(groupID),
  654. "model", requestedModel,
  655. "restricted_accounts", channelRestrictedCount,
  656. "total_accounts", len(accounts))
  657. return nil, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
  658. }
  659. return nil, ErrNoAvailableAccounts
  660. }
  661. accountLoads := make([]AccountWithConcurrency, 0, len(candidates))
  662. for _, acc := range candidates {
  663. accountLoads = append(accountLoads, AccountWithConcurrency{
  664. ID: acc.ID,
  665. MaxConcurrency: acc.EffectiveLoadFactor(),
  666. })
  667. }
  668. loadMap, err := s.concurrencyService.GetAccountsLoadBatch(ctx, accountLoads)
  669. if err != nil {
  670. if result, ok, legacyErr := s.tryAcquireByLegacyOrder(ctx, candidates, groupID, sessionHash, preferOAuth); legacyErr != nil {
  671. return nil, legacyErr
  672. } else if ok {
  673. return result, nil
  674. }
  675. } else {
  676. var available []accountWithLoad
  677. for _, acc := range candidates {
  678. loadInfo := loadMap[acc.ID]
  679. if loadInfo == nil {
  680. loadInfo = &AccountLoadInfo{AccountID: acc.ID}
  681. }
  682. if loadInfo.LoadRate < 100 {
  683. available = append(available, accountWithLoad{
  684. account: acc,
  685. loadInfo: loadInfo,
  686. })
  687. }
  688. }
  689. // 分层过滤选择:优先级 →(可选)最早重置 → 负载率 → LRU
  690. for len(available) > 0 {
  691. // 1. 取优先级最小的集合
  692. candidates := filterByMinPriority(available)
  693. // 2. (可选)use-it-or-lose-it:优先选用会话窗口最早重置的账号
  694. if cfg.PreferSoonestReset {
  695. candidates = filterBySoonestReset(candidates)
  696. }
  697. // 3. 取负载率最低的集合
  698. candidates = filterByMinLoadRate(candidates)
  699. // 4. LRU 选择最久未用的账号
  700. selected := selectByLRU(candidates, preferOAuth)
  701. if selected == nil {
  702. break
  703. }
  704. result, err := s.tryAcquireAccountSlot(ctx, selected.account.ID, selected.account.Concurrency)
  705. if err == nil && result.Acquired {
  706. // 会话数量限制检查
  707. if !s.checkAndRegisterSession(ctx, selected.account, sessionHash) {
  708. result.ReleaseFunc() // 释放槽位,继续尝试下一个账号
  709. } else {
  710. if sessionHash != "" && s.cache != nil {
  711. _ = s.bindGatewayStickySessionDuringSelection(ctx, groupID, sessionHash, selected.account.ID)
  712. }
  713. return s.newSelectionResult(ctx, selected.account, true, result.ReleaseFunc, nil)
  714. }
  715. }
  716. // 移除已尝试的账号,重新进行分层过滤
  717. selectedID := selected.account.ID
  718. newAvailable := make([]accountWithLoad, 0, len(available)-1)
  719. for _, acc := range available {
  720. if acc.account.ID != selectedID {
  721. newAvailable = append(newAvailable, acc)
  722. }
  723. }
  724. available = newAvailable
  725. }
  726. }
  727. // ============ Layer 3: 兜底排队 ============
  728. s.sortCandidatesForFallback(candidates, preferOAuth, cfg.FallbackSelectionMode)
  729. for _, acc := range candidates {
  730. // 会话数量限制检查(等待计划也需要占用会话配额)
  731. if !s.checkAndRegisterSession(ctx, acc, sessionHash) {
  732. continue // 会话限制已满,尝试下一个账号
  733. }
  734. return s.newSelectionResult(ctx, acc, false, nil, &AccountWaitPlan{
  735. AccountID: acc.ID,
  736. MaxConcurrency: acc.Concurrency,
  737. Timeout: cfg.FallbackWaitTimeout,
  738. MaxWaiting: cfg.FallbackMaxWaiting,
  739. })
  740. }
  741. return nil, ErrNoAvailableAccounts
  742. }
  743. func (s *GatewayService) tryAcquireByLegacyOrder(ctx context.Context, candidates []*Account, groupID *int64, sessionHash string, preferOAuth bool) (*AccountSelectionResult, bool, error) {
  744. ordered := append([]*Account(nil), candidates...)
  745. sortAccountsByPriorityAndLastUsed(ordered, preferOAuth)
  746. for _, acc := range ordered {
  747. result, err := s.tryAcquireAccountSlot(ctx, acc.ID, acc.Concurrency)
  748. if err == nil && result.Acquired {
  749. // 会话数量限制检查
  750. if !s.checkAndRegisterSession(ctx, acc, sessionHash) {
  751. result.ReleaseFunc() // 释放槽位,继续尝试下一个账号
  752. continue
  753. }
  754. if sessionHash != "" && s.cache != nil {
  755. _ = s.bindGatewayStickySessionDuringSelection(ctx, groupID, sessionHash, acc.ID)
  756. }
  757. selection, err := s.newSelectionResult(ctx, acc, true, result.ReleaseFunc, nil)
  758. if err != nil {
  759. return nil, false, err
  760. }
  761. return selection, true, nil
  762. }
  763. }
  764. return nil, false, nil
  765. }
  766. func (s *GatewayService) schedulingConfig() config.GatewaySchedulingConfig {
  767. if s.cfg != nil {
  768. return s.cfg.Gateway.Scheduling
  769. }
  770. return config.GatewaySchedulingConfig{
  771. StickySessionMaxWaiting: 3,
  772. StickySessionWaitTimeout: 45 * time.Second,
  773. FallbackWaitTimeout: 30 * time.Second,
  774. FallbackMaxWaiting: 100,
  775. LoadBatchEnabled: true,
  776. SlotCleanupInterval: 30 * time.Second,
  777. }
  778. }
  779. func (s *GatewayService) withGroupContext(ctx context.Context, group *Group) context.Context {
  780. if !IsGroupContextValid(group) {
  781. return ctx
  782. }
  783. if existing, ok := ctx.Value(ctxkey.Group).(*Group); ok && existing != nil && existing.ID == group.ID && IsGroupContextValid(existing) {
  784. return ctx
  785. }
  786. return context.WithValue(ctx, ctxkey.Group, group)
  787. }
  788. func (s *GatewayService) groupFromContext(ctx context.Context, groupID int64) *Group {
  789. if group, ok := ctx.Value(ctxkey.Group).(*Group); ok && IsGroupContextValid(group) && group.ID == groupID {
  790. return group
  791. }
  792. return nil
  793. }
  794. func (s *GatewayService) resolveGroupByID(ctx context.Context, groupID int64) (*Group, error) {
  795. if group := s.groupFromContext(ctx, groupID); group != nil {
  796. return group, nil
  797. }
  798. group, err := s.groupRepo.GetByIDLite(ctx, groupID)
  799. if err != nil {
  800. return nil, fmt.Errorf("get group failed: %w", err)
  801. }
  802. return group, nil
  803. }
  804. func (s *GatewayService) ResolveGroupByID(ctx context.Context, groupID int64) (*Group, error) {
  805. return s.resolveGroupByID(ctx, groupID)
  806. }
  807. // modelRoutingAppliesToPlatform 判定模型路由规则是否适用于本次请求。
  808. //
  809. // 路由规则的存储与查表(Group.ModelRouting / GetRoutingAccountIDs)本身与平台无关,
  810. // 早期只有 Anthropic 目标平台会读取它。OpenAI 分组同样存在“同一个公开别名由不同账号
  811. // 映射到不同上游模型”的故障转移写法,需要显式路由来指定谁是主、谁是备,因此这里把
  812. // Anthropic 与 OpenAI 一并放行。
  813. //
  814. // targetPlatform 是请求解析后的目标平台;groupPlatform 是分组自身的平台。composite
  815. // 分组不限定自身平台,只要目标平台落在放行集合内即可复用其规则。
  816. func modelRoutingAppliesToPlatform(targetPlatform, groupPlatform string) bool {
  817. if !modelRoutingAppliesToTargetPlatform(targetPlatform) {
  818. return false
  819. }
  820. return groupPlatform == targetPlatform || groupPlatform == PlatformComposite
  821. }
  822. // modelRoutingAppliesToTargetPlatform 是放行平台集合的唯一定义处:新增平台只改这里。
  823. // 在只知道目标平台、还没取到分组的位置(调试日志、取分组前的短路)单独判定用。
  824. func modelRoutingAppliesToTargetPlatform(targetPlatform string) bool {
  825. switch targetPlatform {
  826. case PlatformAnthropic, PlatformOpenAI:
  827. return true
  828. default:
  829. return false
  830. }
  831. }
  832. func (s *GatewayService) routingAccountIDsForRequest(ctx context.Context, groupID *int64, requestedModel string, platform string) []int64 {
  833. if groupID == nil || requestedModel == "" || !modelRoutingAppliesToTargetPlatform(platform) {
  834. return nil
  835. }
  836. group, err := s.resolveGroupByID(ctx, *groupID)
  837. if err != nil || group == nil {
  838. if s.debugModelRoutingEnabled() {
  839. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] resolve group failed: group_id=%v model=%s platform=%s err=%v", derefGroupID(groupID), requestedModel, platform, err)
  840. }
  841. return nil
  842. }
  843. // 路由规则适用于解析到 Anthropic 或 OpenAI 的请求;composite 分组在其模型解析到
  844. // 上述平台后同样可以复用这些规则。
  845. if !modelRoutingAppliesToPlatform(platform, group.Platform) {
  846. if s.debugModelRoutingEnabled() {
  847. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] skip: group platform not eligible: group_id=%d group_platform=%s target_platform=%s model=%s", group.ID, group.Platform, platform, requestedModel)
  848. }
  849. return nil
  850. }
  851. ids := group.GetRoutingAccountIDs(requestedModel)
  852. if s.debugModelRoutingEnabled() {
  853. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] routing lookup: group_id=%d model=%s enabled=%v rules=%d matched_ids=%v",
  854. group.ID, requestedModel, group.ModelRoutingEnabled, len(group.ModelRouting), ids)
  855. }
  856. return ids
  857. }
  858. func (s *GatewayService) resolveGatewayGroup(ctx context.Context, groupID *int64) (*Group, *int64, error) {
  859. if groupID == nil {
  860. return nil, nil, nil
  861. }
  862. currentID := *groupID
  863. visited := map[int64]struct{}{}
  864. for {
  865. if _, seen := visited[currentID]; seen {
  866. return nil, nil, fmt.Errorf("fallback group cycle detected")
  867. }
  868. visited[currentID] = struct{}{}
  869. group, err := s.resolveGroupByID(ctx, currentID)
  870. if err != nil {
  871. return nil, nil, err
  872. }
  873. if !group.ClaudeCodeOnly || IsClaudeCodeClient(ctx) {
  874. return group, &currentID, nil
  875. }
  876. if group.FallbackGroupID == nil {
  877. return nil, nil, ErrClaudeCodeOnly
  878. }
  879. currentID = *group.FallbackGroupID
  880. }
  881. }
  882. // checkClaudeCodeRestriction 检查分组的 Claude Code 客户端限制
  883. // 如果分组启用了 claude_code_only 且请求不是来自 Claude Code 客户端:
  884. // - 有降级分组:返回降级分组的 ID
  885. // - 无降级分组:返回 ErrClaudeCodeOnly 错误
  886. func (s *GatewayService) checkClaudeCodeRestriction(ctx context.Context, groupID *int64) (*Group, *int64, error) {
  887. if groupID == nil {
  888. return nil, groupID, nil
  889. }
  890. // 强制平台模式不检查 Claude Code 限制
  891. if forcePlatform, hasForcePlatform := ctx.Value(ctxkey.ForcePlatform).(string); hasForcePlatform && forcePlatform != "" {
  892. return nil, groupID, nil
  893. }
  894. group, resolvedID, err := s.resolveGatewayGroup(ctx, groupID)
  895. if err != nil {
  896. return nil, nil, err
  897. }
  898. return group, resolvedID, nil
  899. }
  900. func (s *GatewayService) resolvePlatform(ctx context.Context, groupID *int64, group *Group, requestedModel string) (string, bool, error) {
  901. forcePlatform, hasForcePlatform := ctx.Value(ctxkey.ForcePlatform).(string)
  902. if hasForcePlatform && forcePlatform != "" {
  903. return forcePlatform, true, nil
  904. }
  905. if platform, ok := ResolvedTargetPlatformFromContext(ctx); ok {
  906. return platform, false, nil
  907. }
  908. if group != nil {
  909. if group.Platform == PlatformComposite {
  910. decision, ok, err := s.resolveCompositeRouteDecision(ctx, group, requestedModel, CompositeRouteEndpointAny)
  911. if err != nil {
  912. return "", false, err
  913. }
  914. if !ok {
  915. return "", false, fmt.Errorf("%w supporting model: %s (composite target platform unknown)", ErrNoAvailableAccounts, requestedModel)
  916. }
  917. return decision.TargetPlatform, false, nil
  918. }
  919. return group.Platform, false, nil
  920. }
  921. if groupID != nil {
  922. group, err := s.resolveGroupByID(ctx, *groupID)
  923. if err != nil {
  924. return "", false, err
  925. }
  926. if group.Platform == PlatformComposite {
  927. decision, ok, err := s.resolveCompositeRouteDecision(ctx, group, requestedModel, CompositeRouteEndpointAny)
  928. if err != nil {
  929. return "", false, err
  930. }
  931. if !ok {
  932. return "", false, fmt.Errorf("%w supporting model: %s (composite target platform unknown)", ErrNoAvailableAccounts, requestedModel)
  933. }
  934. return decision.TargetPlatform, false, nil
  935. }
  936. return group.Platform, false, nil
  937. }
  938. return PlatformAnthropic, false, nil
  939. }
  940. func (s *GatewayService) listSchedulableAccounts(ctx context.Context, groupID *int64, platform string, hasForcePlatform bool) ([]Account, bool, error) {
  941. if s.schedulerSnapshot != nil {
  942. accounts, useMixed, err := s.schedulerSnapshot.ListSchedulableAccounts(ctx, groupID, platform, hasForcePlatform)
  943. if err == nil {
  944. accounts = s.filterAccountsBySchedulingThreshold(ctx, accounts)
  945. if platform == PlatformGrok || strings.EqualFold(platform, PlatformGrok) {
  946. accounts = s.filterGrokFreeQuotaAccountsForGateway(ctx, accounts)
  947. }
  948. slog.Debug("account_scheduling_list_snapshot",
  949. "group_id", derefGroupID(groupID),
  950. "platform", platform,
  951. "use_mixed", useMixed,
  952. "count", len(accounts))
  953. if slog.Default().Enabled(ctx, slog.LevelDebug) {
  954. for _, acc := range accounts {
  955. slog.Debug("account_scheduling_account_detail",
  956. "account_id", acc.ID,
  957. "name", acc.Name,
  958. "platform", acc.Platform,
  959. "type", acc.Type,
  960. "status", acc.Status,
  961. "tls_fingerprint", acc.IsTLSFingerprintEnabled())
  962. }
  963. }
  964. }
  965. if err == nil {
  966. accounts = FilterTierPoolAccounts(accounts, TierPoolFromContext(ctx))
  967. }
  968. return accounts, useMixed, err
  969. }
  970. useMixed := (platform == PlatformAnthropic || platform == PlatformGemini) && !hasForcePlatform
  971. if useMixed {
  972. platforms := []string{platform, PlatformAntigravity}
  973. var accounts []Account
  974. var err error
  975. if groupID != nil {
  976. accounts, err = s.accountRepo.ListSchedulableByGroupIDAndPlatforms(ctx, *groupID, platforms)
  977. } else if s.cfg != nil && s.cfg.RunMode == config.RunModeSimple {
  978. accounts, err = s.accountRepo.ListSchedulableByPlatforms(ctx, platforms)
  979. } else {
  980. accounts, err = s.accountRepo.ListSchedulableUngroupedByPlatforms(ctx, platforms)
  981. }
  982. if err != nil {
  983. slog.Debug("account_scheduling_list_failed",
  984. "group_id", derefGroupID(groupID),
  985. "platform", platform,
  986. "error", err)
  987. return nil, useMixed, err
  988. }
  989. filtered := make([]Account, 0, len(accounts))
  990. for _, acc := range accounts {
  991. if acc.Platform == PlatformAntigravity && !acc.IsMixedSchedulingEnabled() {
  992. continue
  993. }
  994. filtered = append(filtered, acc)
  995. }
  996. slog.Debug("account_scheduling_list_mixed",
  997. "group_id", derefGroupID(groupID),
  998. "platform", platform,
  999. "raw_count", len(accounts),
  1000. "filtered_count", len(filtered))
  1001. if slog.Default().Enabled(ctx, slog.LevelDebug) {
  1002. for _, acc := range filtered {
  1003. slog.Debug("account_scheduling_account_detail",
  1004. "account_id", acc.ID,
  1005. "name", acc.Name,
  1006. "platform", acc.Platform,
  1007. "type", acc.Type,
  1008. "status", acc.Status,
  1009. "tls_fingerprint", acc.IsTLSFingerprintEnabled())
  1010. }
  1011. }
  1012. return FilterTierPoolAccounts(s.filterAccountsBySchedulingThreshold(ctx, filtered), TierPoolFromContext(ctx)), useMixed, nil
  1013. }
  1014. var accounts []Account
  1015. var err error
  1016. if s.cfg != nil && s.cfg.RunMode == config.RunModeSimple {
  1017. accounts, err = s.accountRepo.ListSchedulableByPlatform(ctx, platform)
  1018. } else if groupID != nil {
  1019. accounts, err = s.accountRepo.ListSchedulableByGroupIDAndPlatform(ctx, *groupID, platform)
  1020. // 分组内无账号则返回空列表,由上层处理错误,不再回退到全平台查询
  1021. } else {
  1022. accounts, err = s.accountRepo.ListSchedulableUngroupedByPlatform(ctx, platform)
  1023. }
  1024. if err != nil {
  1025. slog.Debug("account_scheduling_list_failed",
  1026. "group_id", derefGroupID(groupID),
  1027. "platform", platform,
  1028. "error", err)
  1029. return nil, useMixed, err
  1030. }
  1031. slog.Debug("account_scheduling_list_single",
  1032. "group_id", derefGroupID(groupID),
  1033. "platform", platform,
  1034. "count", len(accounts))
  1035. if slog.Default().Enabled(ctx, slog.LevelDebug) {
  1036. for _, acc := range accounts {
  1037. slog.Debug("account_scheduling_account_detail",
  1038. "account_id", acc.ID,
  1039. "name", acc.Name,
  1040. "platform", acc.Platform,
  1041. "type", acc.Type,
  1042. "status", acc.Status,
  1043. "tls_fingerprint", acc.IsTLSFingerprintEnabled())
  1044. }
  1045. }
  1046. accounts = s.filterAccountsBySchedulingThreshold(ctx, accounts)
  1047. if platform == PlatformGrok || strings.EqualFold(platform, PlatformGrok) {
  1048. accounts = s.filterGrokFreeQuotaAccountsForGateway(ctx, accounts)
  1049. }
  1050. return FilterTierPoolAccounts(accounts, TierPoolFromContext(ctx)), useMixed, nil
  1051. }
  1052. // IsSingleAntigravityAccountGroup 检查指定分组是否只有一个 antigravity 平台的可调度账号。
  1053. // 用于 Handler 层在首次请求时提前设置 SingleAccountRetry context,
  1054. // 避免单账号分组收到 503 时错误地设置模型限流标记导致后续请求连续快速失败。
  1055. func (s *GatewayService) IsSingleAntigravityAccountGroup(ctx context.Context, groupID *int64) bool {
  1056. accounts, _, err := s.listSchedulableAccounts(ctx, groupID, PlatformAntigravity, true)
  1057. if err != nil {
  1058. return false
  1059. }
  1060. return len(accounts) == 1
  1061. }
  1062. func (s *GatewayService) isAccountAllowedForPlatform(account *Account, platform string, useMixed bool) bool {
  1063. if account == nil {
  1064. return false
  1065. }
  1066. if useMixed {
  1067. if account.Platform == platform {
  1068. return true
  1069. }
  1070. return account.Platform == PlatformAntigravity && account.IsMixedSchedulingEnabled()
  1071. }
  1072. return account.Platform == platform
  1073. }
  1074. func (s *GatewayService) isAccountSchedulableForSelection(account *Account) bool {
  1075. if account == nil {
  1076. return false
  1077. }
  1078. return account.IsSchedulable()
  1079. }
  1080. func (s *GatewayService) isAccountSchedulableForModelSelection(ctx context.Context, account *Account, requestedModel string) bool {
  1081. if account == nil {
  1082. return false
  1083. }
  1084. return account.IsSchedulableForModelWithContext(ctx, requestedModel)
  1085. }
  1086. // isAccountInGroup checks if the account belongs to the specified group.
  1087. // When groupID is nil, returns true only for ungrouped accounts (no group assignments).
  1088. func (s *GatewayService) isAccountInGroup(account *Account, groupID *int64) bool {
  1089. if account == nil {
  1090. return false
  1091. }
  1092. if groupID == nil {
  1093. // 无分组的 API Key 只能使用未分组的账号
  1094. return len(account.AccountGroups) == 0
  1095. }
  1096. for _, ag := range account.AccountGroups {
  1097. if ag.GroupID == *groupID {
  1098. return true
  1099. }
  1100. }
  1101. return false
  1102. }
  1103. func (s *GatewayService) tryAcquireAccountSlot(ctx context.Context, accountID int64, maxConcurrency int) (*AcquireResult, error) {
  1104. if s.concurrencyService == nil {
  1105. return &AcquireResult{Acquired: true, ReleaseFunc: func() {}}, nil
  1106. }
  1107. return s.concurrencyService.AcquireAccountSlot(ctx, accountID, maxConcurrency)
  1108. }
  1109. type usageLogWindowStatsBatchProvider interface {
  1110. GetAccountWindowStatsBatch(ctx context.Context, accountIDs []int64, startTime time.Time) (map[int64]*usagestats.AccountStats, error)
  1111. }
  1112. type windowCostPrefetchContextKeyType struct{}
  1113. var windowCostPrefetchContextKey = windowCostPrefetchContextKeyType{}
  1114. func windowCostFromPrefetchContext(ctx context.Context, accountID int64) (float64, bool) {
  1115. if ctx == nil || accountID <= 0 {
  1116. return 0, false
  1117. }
  1118. m, ok := ctx.Value(windowCostPrefetchContextKey).(map[int64]float64)
  1119. if !ok || len(m) == 0 {
  1120. return 0, false
  1121. }
  1122. v, exists := m[accountID]
  1123. return v, exists
  1124. }
  1125. func (s *GatewayService) withWindowCostPrefetch(ctx context.Context, accounts []Account) context.Context {
  1126. if ctx == nil || len(accounts) == 0 || s.sessionLimitCache == nil || s.usageLogRepo == nil {
  1127. return ctx
  1128. }
  1129. accountByID := make(map[int64]*Account)
  1130. accountIDs := make([]int64, 0, len(accounts))
  1131. for i := range accounts {
  1132. account := &accounts[i]
  1133. if account == nil || !account.IsAnthropicOAuthOrSetupToken() {
  1134. continue
  1135. }
  1136. if account.GetWindowCostLimit() <= 0 {
  1137. continue
  1138. }
  1139. accountByID[account.ID] = account
  1140. accountIDs = append(accountIDs, account.ID)
  1141. }
  1142. if len(accountIDs) == 0 {
  1143. return ctx
  1144. }
  1145. costs := make(map[int64]float64, len(accountIDs))
  1146. cacheValues, err := s.sessionLimitCache.GetWindowCostBatch(ctx, accountIDs)
  1147. if err == nil {
  1148. for accountID, cost := range cacheValues {
  1149. costs[accountID] = cost
  1150. }
  1151. windowCostPrefetchCacheHitTotal.Add(int64(len(cacheValues)))
  1152. } else {
  1153. windowCostPrefetchErrorTotal.Add(1)
  1154. logger.LegacyPrintf("service.gateway", "window_cost batch cache read failed: %v", err)
  1155. }
  1156. cacheMissCount := len(accountIDs) - len(costs)
  1157. if cacheMissCount < 0 {
  1158. cacheMissCount = 0
  1159. }
  1160. windowCostPrefetchCacheMissTotal.Add(int64(cacheMissCount))
  1161. missingByStart := make(map[int64][]int64)
  1162. startTimes := make(map[int64]time.Time)
  1163. for _, accountID := range accountIDs {
  1164. if _, ok := costs[accountID]; ok {
  1165. continue
  1166. }
  1167. account := accountByID[accountID]
  1168. if account == nil {
  1169. continue
  1170. }
  1171. startTime := account.GetCurrentWindowStartTime()
  1172. startKey := startTime.Unix()
  1173. missingByStart[startKey] = append(missingByStart[startKey], accountID)
  1174. startTimes[startKey] = startTime
  1175. }
  1176. if len(missingByStart) == 0 {
  1177. return context.WithValue(ctx, windowCostPrefetchContextKey, costs)
  1178. }
  1179. batchReader, hasBatch := s.usageLogRepo.(usageLogWindowStatsBatchProvider)
  1180. for startKey, ids := range missingByStart {
  1181. startTime := startTimes[startKey]
  1182. if hasBatch {
  1183. windowCostPrefetchBatchSQLTotal.Add(1)
  1184. queryStart := time.Now()
  1185. statsByAccount, err := batchReader.GetAccountWindowStatsBatch(ctx, ids, startTime)
  1186. if err == nil {
  1187. slog.Debug("window_cost_batch_query_ok",
  1188. "accounts", len(ids),
  1189. "window_start", startTime.Format(time.RFC3339),
  1190. "duration_ms", time.Since(queryStart).Milliseconds())
  1191. for _, accountID := range ids {
  1192. stats := statsByAccount[accountID]
  1193. cost := 0.0
  1194. if stats != nil {
  1195. cost = stats.StandardCost
  1196. }
  1197. costs[accountID] = cost
  1198. _ = s.sessionLimitCache.SetWindowCost(ctx, accountID, cost)
  1199. }
  1200. continue
  1201. }
  1202. windowCostPrefetchErrorTotal.Add(1)
  1203. logger.LegacyPrintf("service.gateway", "window_cost batch db query failed: start=%s err=%v", startTime.Format(time.RFC3339), err)
  1204. }
  1205. // 回退路径:缺少批量仓储能力或批量查询失败时,按账号单查(失败开放)。
  1206. windowCostPrefetchFallbackTotal.Add(int64(len(ids)))
  1207. for _, accountID := range ids {
  1208. stats, err := s.usageLogRepo.GetAccountWindowStats(ctx, accountID, startTime)
  1209. if err != nil {
  1210. windowCostPrefetchErrorTotal.Add(1)
  1211. continue
  1212. }
  1213. cost := stats.StandardCost
  1214. costs[accountID] = cost
  1215. _ = s.sessionLimitCache.SetWindowCost(ctx, accountID, cost)
  1216. }
  1217. }
  1218. return context.WithValue(ctx, windowCostPrefetchContextKey, costs)
  1219. }
  1220. // isAccountSchedulableForQuota 检查账号是否在配额限制内
  1221. // 适用于配置了 quota_limit 的 apikey 和 bedrock 类型账号
  1222. func (s *GatewayService) isAccountSchedulableForQuota(account *Account) bool {
  1223. if !account.IsAPIKeyOrBedrock() {
  1224. return true
  1225. }
  1226. return !account.IsQuotaExceeded()
  1227. }
  1228. // isAccountSchedulableForWindowCost 检查账号是否可根据窗口费用进行调度
  1229. // 仅适用于 Anthropic OAuth/SetupToken 账号
  1230. // 返回 true 表示可调度,false 表示不可调度
  1231. func (s *GatewayService) isAccountSchedulableForWindowCost(ctx context.Context, account *Account, isSticky bool) bool {
  1232. // 只检查 Anthropic OAuth/SetupToken 账号
  1233. if !account.IsAnthropicOAuthOrSetupToken() {
  1234. return true
  1235. }
  1236. limit := account.GetWindowCostLimit()
  1237. if limit <= 0 {
  1238. return true // 未启用窗口费用限制
  1239. }
  1240. // 尝试从缓存获取窗口费用
  1241. var currentCost float64
  1242. if cost, ok := windowCostFromPrefetchContext(ctx, account.ID); ok {
  1243. currentCost = cost
  1244. goto checkSchedulability
  1245. }
  1246. if s.sessionLimitCache != nil {
  1247. if cost, hit, err := s.sessionLimitCache.GetWindowCost(ctx, account.ID); err == nil && hit {
  1248. currentCost = cost
  1249. goto checkSchedulability
  1250. }
  1251. }
  1252. // 缓存未命中,从数据库查询
  1253. {
  1254. // 使用统一的窗口开始时间计算逻辑(考虑窗口过期情况)
  1255. startTime := account.GetCurrentWindowStartTime()
  1256. stats, err := s.usageLogRepo.GetAccountWindowStats(ctx, account.ID, startTime)
  1257. if err != nil {
  1258. // 失败开放:查询失败时允许调度
  1259. return true
  1260. }
  1261. // 使用标准费用(不含账号倍率)
  1262. currentCost = stats.StandardCost
  1263. // 设置缓存(忽略错误)
  1264. if s.sessionLimitCache != nil {
  1265. _ = s.sessionLimitCache.SetWindowCost(ctx, account.ID, currentCost)
  1266. }
  1267. }
  1268. checkSchedulability:
  1269. schedulability := account.CheckWindowCostSchedulability(currentCost)
  1270. switch schedulability {
  1271. case WindowCostSchedulable:
  1272. return true
  1273. case WindowCostStickyOnly:
  1274. return isSticky
  1275. case WindowCostNotSchedulable:
  1276. return false
  1277. }
  1278. return true
  1279. }
  1280. // rpmPrefetchContextKey is the context key for prefetched RPM counts.
  1281. type rpmPrefetchContextKeyType struct{}
  1282. var rpmPrefetchContextKey = rpmPrefetchContextKeyType{}
  1283. func rpmFromPrefetchContext(ctx context.Context, accountID int64) (int, bool) {
  1284. if v, ok := ctx.Value(rpmPrefetchContextKey).(map[int64]int); ok {
  1285. count, found := v[accountID]
  1286. return count, found
  1287. }
  1288. return 0, false
  1289. }
  1290. // withRPMPrefetch 批量预取所有候选账号的 RPM 计数
  1291. func (s *GatewayService) withRPMPrefetch(ctx context.Context, accounts []Account) context.Context {
  1292. if s.rpmCache == nil {
  1293. return ctx
  1294. }
  1295. var ids []int64
  1296. for i := range accounts {
  1297. if accounts[i].IsAnthropicOAuthOrSetupToken() && accounts[i].GetBaseRPM() > 0 {
  1298. ids = append(ids, accounts[i].ID)
  1299. }
  1300. }
  1301. if len(ids) == 0 {
  1302. return ctx
  1303. }
  1304. counts, err := s.rpmCache.GetRPMBatch(ctx, ids)
  1305. if err != nil {
  1306. return ctx // 失败开放
  1307. }
  1308. return context.WithValue(ctx, rpmPrefetchContextKey, counts)
  1309. }
  1310. // isAccountSchedulableForRPM 检查账号是否可根据 RPM 进行调度
  1311. // 仅适用于 Anthropic OAuth/SetupToken 账号
  1312. func (s *GatewayService) isAccountSchedulableForRPM(ctx context.Context, account *Account, isSticky bool) bool {
  1313. if !account.IsAnthropicOAuthOrSetupToken() {
  1314. return true
  1315. }
  1316. baseRPM := account.GetBaseRPM()
  1317. if baseRPM <= 0 {
  1318. return true
  1319. }
  1320. // 尝试从预取缓存获取
  1321. var currentRPM int
  1322. if count, ok := rpmFromPrefetchContext(ctx, account.ID); ok {
  1323. currentRPM = count
  1324. } else if s.rpmCache != nil {
  1325. if count, err := s.rpmCache.GetRPM(ctx, account.ID); err == nil {
  1326. currentRPM = count
  1327. }
  1328. // 失败开放:GetRPM 错误时允许调度
  1329. }
  1330. schedulability := account.CheckRPMSchedulability(currentRPM)
  1331. switch schedulability {
  1332. case WindowCostSchedulable:
  1333. return true
  1334. case WindowCostStickyOnly:
  1335. return isSticky
  1336. case WindowCostNotSchedulable:
  1337. return false
  1338. }
  1339. return true
  1340. }
  1341. // IncrementAccountRPM increments the RPM counter for the given account.
  1342. // 已知 TOCTOU 竞态:调度时读取 RPM 计数与此处递增之间存在时间窗口,
  1343. // 高并发下可能短暂超出 RPM 限制。这是与 WindowCost 一致的 soft-limit
  1344. // 设计权衡——可接受的少量超额优于加锁带来的延迟和复杂度。
  1345. func (s *GatewayService) IncrementAccountRPM(ctx context.Context, accountID int64) error {
  1346. if s.rpmCache == nil {
  1347. return nil
  1348. }
  1349. _, err := s.rpmCache.IncrementRPM(ctx, accountID)
  1350. return err
  1351. }
  1352. // checkAndRegisterSession 检查并注册会话,用于会话数量限制
  1353. // 仅适用于 Anthropic OAuth/SetupToken 账号
  1354. // sessionID: 会话标识符(使用粘性会话的 hash)
  1355. // 返回 true 表示允许(在限制内或会话已存在),false 表示拒绝(超出限制且是新会话)
  1356. func (s *GatewayService) checkAndRegisterSession(ctx context.Context, account *Account, sessionID string) bool {
  1357. // 只检查 Anthropic OAuth/SetupToken 账号
  1358. if !account.IsAnthropicOAuthOrSetupToken() {
  1359. return true
  1360. }
  1361. maxSessions := account.GetMaxSessions()
  1362. if maxSessions <= 0 || sessionID == "" {
  1363. return true // 未启用会话限制或无会话ID
  1364. }
  1365. if s.sessionLimitCache == nil {
  1366. return true // 缓存不可用时允许通过
  1367. }
  1368. idleTimeout := time.Duration(account.GetSessionIdleTimeoutMinutes()) * time.Minute
  1369. allowed, err := s.sessionLimitCache.RegisterSession(ctx, account.ID, sessionID, maxSessions, idleTimeout)
  1370. if err != nil {
  1371. // 失败开放:缓存错误时允许通过
  1372. return true
  1373. }
  1374. return allowed
  1375. }
  1376. // ReleaseAccountSession 立即释放会话槽(不等待空闲超时)
  1377. // 供 handler 在请求最终失败(选号成功但转发失败/客户端中断)时调用:
  1378. // 上游从未真正服务该会话,若继续占槽,max_sessions 受限的账号会被失败请求的
  1379. // session hash 卡满整个空闲窗口,后续新会话全部被拒。
  1380. // 适用条件与 checkAndRegisterSession 对齐;不适用账号为 no-op,幂等可安全重复调用。
  1381. func (s *GatewayService) ReleaseAccountSession(ctx context.Context, account *Account, sessionID string) {
  1382. if s == nil || s.sessionLimitCache == nil || account == nil || sessionID == "" {
  1383. return
  1384. }
  1385. if !account.IsAnthropicOAuthOrSetupToken() {
  1386. return
  1387. }
  1388. if account.GetMaxSessions() <= 0 {
  1389. return
  1390. }
  1391. if err := s.sessionLimitCache.UnregisterSession(ctx, account.ID, sessionID); err != nil {
  1392. slog.Debug("session_limit.release_failed",
  1393. "account_id", account.ID,
  1394. "error", err)
  1395. }
  1396. }
  1397. func (s *GatewayService) getSchedulableAccount(ctx context.Context, accountID int64) (*Account, error) {
  1398. var (
  1399. account *Account
  1400. err error
  1401. )
  1402. if s.schedulerSnapshot != nil {
  1403. account, err = s.schedulerSnapshot.GetAccount(ctx, accountID)
  1404. } else {
  1405. account, err = s.accountRepo.GetByID(ctx, accountID)
  1406. }
  1407. if err != nil || account == nil {
  1408. return account, err
  1409. }
  1410. if s.isAccountBlockedBySchedulingThreshold(ctx, account) {
  1411. return nil, nil
  1412. }
  1413. // Sticky / non-list selection must honor free soft-gate (same as listSchedulableAccounts).
  1414. if account.IsGrok() {
  1415. if gated := s.filterGrokFreeQuotaAccountsForGateway(ctx, []Account{*account}); len(gated) == 0 {
  1416. return nil, nil
  1417. }
  1418. }
  1419. return account, nil
  1420. }
  1421. func (s *GatewayService) filterAccountsBySchedulingThreshold(ctx context.Context, accounts []Account) []Account {
  1422. if len(accounts) == 0 {
  1423. return accounts
  1424. }
  1425. filtered := make([]Account, 0, len(accounts))
  1426. for i := range accounts {
  1427. if s.isAccountBlockedBySchedulingThreshold(ctx, &accounts[i]) {
  1428. continue
  1429. }
  1430. filtered = append(filtered, accounts[i])
  1431. }
  1432. return filtered
  1433. }
  1434. func (s *GatewayService) isAccountBlockedBySchedulingThreshold(ctx context.Context, account *Account) bool {
  1435. if s == nil || s.rateLimitService == nil || account == nil {
  1436. return false
  1437. }
  1438. return s.rateLimitService.ApplyAccountSchedulingThreshold(ctx, account)
  1439. }
  1440. func (s *GatewayService) hydrateSelectedAccount(ctx context.Context, account *Account) (*Account, error) {
  1441. if account == nil {
  1442. return nil, nil
  1443. }
  1444. hydrated := account
  1445. if s.schedulerSnapshot != nil {
  1446. var err error
  1447. hydrated, err = s.schedulerSnapshot.GetAccount(ctx, account.ID)
  1448. if err != nil {
  1449. return nil, err
  1450. }
  1451. if hydrated == nil {
  1452. return nil, fmt.Errorf("selected gateway account %d not found during hydration", account.ID)
  1453. }
  1454. }
  1455. if err := ResolveRandomProxyFromSource(ctx, hydrated, s.accountRepo); err != nil {
  1456. return nil, err
  1457. }
  1458. return hydrated, nil
  1459. }
  1460. func (s *GatewayService) newSelectionResult(ctx context.Context, account *Account, acquired bool, release func(), waitPlan *AccountWaitPlan) (*AccountSelectionResult, error) {
  1461. hydrated, err := s.hydrateSelectedAccount(ctx, account)
  1462. if err != nil {
  1463. return nil, err
  1464. }
  1465. return s.newSelectionResultFromHydratedAccount(ctx, hydrated, acquired, release, waitPlan), nil
  1466. }
  1467. // newSelectionResultFromHydratedAccount builds a result for a caller that has
  1468. // already loaded the full account payload. Keeping this separate from
  1469. // newSelectionResult prevents the no-load scheduling path from fetching the
  1470. // same scheduler snapshot (and selecting a random proxy) twice.
  1471. func (s *GatewayService) newSelectionResultFromHydratedAccount(ctx context.Context, account *Account, acquired bool, release func(), waitPlan *AccountWaitPlan) *AccountSelectionResult {
  1472. if account == nil {
  1473. return nil
  1474. }
  1475. return attachSelectionProfitGate(ctx, &AccountSelectionResult{
  1476. Account: account,
  1477. Acquired: acquired,
  1478. ReleaseFunc: release,
  1479. WaitPlan: waitPlan,
  1480. })
  1481. }
  1482. // filterByMinPriority 过滤出优先级最小的账号集合
  1483. func filterByMinPriority(accounts []accountWithLoad) []accountWithLoad {
  1484. if len(accounts) == 0 {
  1485. return accounts
  1486. }
  1487. minPriority := accounts[0].account.Priority
  1488. for _, acc := range accounts[1:] {
  1489. if acc.account.Priority < minPriority {
  1490. minPriority = acc.account.Priority
  1491. }
  1492. }
  1493. result := make([]accountWithLoad, 0, len(accounts))
  1494. for _, acc := range accounts {
  1495. if acc.account.Priority == minPriority {
  1496. result = append(result, acc)
  1497. }
  1498. }
  1499. return result
  1500. }
  1501. // filterByMinLoadRate 过滤出负载率最低的账号集合
  1502. func filterByMinLoadRate(accounts []accountWithLoad) []accountWithLoad {
  1503. if len(accounts) == 0 {
  1504. return accounts
  1505. }
  1506. minLoadRate := accounts[0].loadInfo.LoadRate
  1507. for _, acc := range accounts[1:] {
  1508. if acc.loadInfo.LoadRate < minLoadRate {
  1509. minLoadRate = acc.loadInfo.LoadRate
  1510. }
  1511. }
  1512. result := make([]accountWithLoad, 0, len(accounts))
  1513. for _, acc := range accounts {
  1514. if acc.loadInfo.LoadRate == minLoadRate {
  1515. result = append(result, acc)
  1516. }
  1517. }
  1518. return result
  1519. }
  1520. // filterBySoonestReset 过滤出「会话窗口最早重置」的账号集合(use-it-or-lose-it)。
  1521. // 仅保留拥有未来重置时间(SessionWindowEnd 在当前时间之后)且最早的账号;
  1522. // 窗口为空或已过期的账号视为无活跃窗口、优先级最低。
  1523. // 当所有账号都没有活跃窗口时,返回原集合(不改变后续 LRU 选择)。
  1524. func filterBySoonestReset(accounts []accountWithLoad) []accountWithLoad {
  1525. if len(accounts) <= 1 {
  1526. return accounts
  1527. }
  1528. now := time.Now()
  1529. var minEnd *time.Time
  1530. for _, acc := range accounts {
  1531. end := acc.account.SessionWindowEnd
  1532. if end == nil || !now.Before(*end) {
  1533. continue
  1534. }
  1535. if minEnd == nil || end.Before(*minEnd) {
  1536. minEnd = end
  1537. }
  1538. }
  1539. if minEnd == nil {
  1540. // 没有任何账号拥有活跃窗口,保持原集合
  1541. return accounts
  1542. }
  1543. result := make([]accountWithLoad, 0, len(accounts))
  1544. for _, acc := range accounts {
  1545. end := acc.account.SessionWindowEnd
  1546. if end != nil && now.Before(*end) && end.Equal(*minEnd) {
  1547. result = append(result, acc)
  1548. }
  1549. }
  1550. return result
  1551. }
  1552. // selectByLRU 从集合中选择最久未用的账号
  1553. // 如果有多个账号具有相同的最小 LastUsedAt,则随机选择一个
  1554. func selectByLRU(accounts []accountWithLoad, preferOAuth bool) *accountWithLoad {
  1555. if len(accounts) == 0 {
  1556. return nil
  1557. }
  1558. if len(accounts) == 1 {
  1559. return &accounts[0]
  1560. }
  1561. // 1. 找到最小的 LastUsedAt(nil 被视为最小)
  1562. var minTime *time.Time
  1563. hasNil := false
  1564. for _, acc := range accounts {
  1565. if acc.account.LastUsedAt == nil {
  1566. hasNil = true
  1567. break
  1568. }
  1569. if minTime == nil || acc.account.LastUsedAt.Before(*minTime) {
  1570. minTime = acc.account.LastUsedAt
  1571. }
  1572. }
  1573. // 2. 收集所有具有最小 LastUsedAt 的账号索引
  1574. var candidateIdxs []int
  1575. for i, acc := range accounts {
  1576. if hasNil {
  1577. if acc.account.LastUsedAt == nil {
  1578. candidateIdxs = append(candidateIdxs, i)
  1579. }
  1580. } else {
  1581. if acc.account.LastUsedAt != nil && acc.account.LastUsedAt.Equal(*minTime) {
  1582. candidateIdxs = append(candidateIdxs, i)
  1583. }
  1584. }
  1585. }
  1586. // 3. 如果只有一个候选,直接返回
  1587. if len(candidateIdxs) == 1 {
  1588. return &accounts[candidateIdxs[0]]
  1589. }
  1590. // 4. 如果有多个候选且 preferOAuth,优先选择 OAuth 类型
  1591. if preferOAuth {
  1592. var oauthIdxs []int
  1593. for _, idx := range candidateIdxs {
  1594. if accounts[idx].account.Type == AccountTypeOAuth {
  1595. oauthIdxs = append(oauthIdxs, idx)
  1596. }
  1597. }
  1598. if len(oauthIdxs) > 0 {
  1599. candidateIdxs = oauthIdxs
  1600. }
  1601. }
  1602. // 5. 随机选择一个
  1603. selectedIdx := candidateIdxs[mathrand.Intn(len(candidateIdxs))]
  1604. return &accounts[selectedIdx]
  1605. }
  1606. func sortAccountsByPriorityAndLastUsed(accounts []*Account, preferOAuth bool) {
  1607. sort.SliceStable(accounts, func(i, j int) bool {
  1608. a, b := accounts[i], accounts[j]
  1609. if a.Priority != b.Priority {
  1610. return a.Priority < b.Priority
  1611. }
  1612. switch {
  1613. case a.LastUsedAt == nil && b.LastUsedAt != nil:
  1614. return true
  1615. case a.LastUsedAt != nil && b.LastUsedAt == nil:
  1616. return false
  1617. case a.LastUsedAt == nil && b.LastUsedAt == nil:
  1618. if preferOAuth && a.Type != b.Type {
  1619. return a.Type == AccountTypeOAuth
  1620. }
  1621. return false
  1622. default:
  1623. return a.LastUsedAt.Before(*b.LastUsedAt)
  1624. }
  1625. })
  1626. shuffleWithinPriorityAndLastUsed(accounts, preferOAuth)
  1627. }
  1628. // shuffleWithinSortGroups 对排序后的 accountWithLoad 切片,按 (Priority, LoadRate, LastUsedAt) 分组后组内随机打乱。
  1629. // 防止并发请求读取同一快照时,确定性排序导致所有请求命中相同账号。
  1630. func shuffleWithinSortGroups(accounts []accountWithLoad) {
  1631. if len(accounts) <= 1 {
  1632. return
  1633. }
  1634. i := 0
  1635. for i < len(accounts) {
  1636. j := i + 1
  1637. for j < len(accounts) && sameAccountWithLoadGroup(accounts[i], accounts[j]) {
  1638. j++
  1639. }
  1640. if j-i > 1 {
  1641. mathrand.Shuffle(j-i, func(a, b int) {
  1642. accounts[i+a], accounts[i+b] = accounts[i+b], accounts[i+a]
  1643. })
  1644. }
  1645. i = j
  1646. }
  1647. }
  1648. // sameAccountWithLoadGroup 判断两个 accountWithLoad 是否属于同一排序组
  1649. func sameAccountWithLoadGroup(a, b accountWithLoad) bool {
  1650. if a.account.Priority != b.account.Priority {
  1651. return false
  1652. }
  1653. if a.loadInfo.LoadRate != b.loadInfo.LoadRate {
  1654. return false
  1655. }
  1656. return sameLastUsedAt(a.account.LastUsedAt, b.account.LastUsedAt)
  1657. }
  1658. // shuffleWithinPriorityAndLastUsed 对排序后的 []*Account 切片,按 (Priority, LastUsedAt) 分组后组内随机打乱。
  1659. //
  1660. // 注意:当 preferOAuth=true 时,需要保证 OAuth 账号在同组内仍然优先,否则会把排序时的偏好打散掉。
  1661. // 因此这里采用"组内分区 + 分区内 shuffle"的方式:
  1662. // - 先把同组账号按 (OAuth / 非 OAuth) 拆成两段,保持 OAuth 段在前;
  1663. // - 再分别在各段内随机打散,避免热点。
  1664. func shuffleWithinPriorityAndLastUsed(accounts []*Account, preferOAuth bool) {
  1665. if len(accounts) <= 1 {
  1666. return
  1667. }
  1668. i := 0
  1669. for i < len(accounts) {
  1670. j := i + 1
  1671. for j < len(accounts) && sameAccountGroup(accounts[i], accounts[j]) {
  1672. j++
  1673. }
  1674. if j-i > 1 {
  1675. if preferOAuth {
  1676. oauth := make([]*Account, 0, j-i)
  1677. others := make([]*Account, 0, j-i)
  1678. for _, acc := range accounts[i:j] {
  1679. if acc.Type == AccountTypeOAuth {
  1680. oauth = append(oauth, acc)
  1681. } else {
  1682. others = append(others, acc)
  1683. }
  1684. }
  1685. if len(oauth) > 1 {
  1686. mathrand.Shuffle(len(oauth), func(a, b int) { oauth[a], oauth[b] = oauth[b], oauth[a] })
  1687. }
  1688. if len(others) > 1 {
  1689. mathrand.Shuffle(len(others), func(a, b int) { others[a], others[b] = others[b], others[a] })
  1690. }
  1691. copy(accounts[i:], oauth)
  1692. copy(accounts[i+len(oauth):], others)
  1693. } else {
  1694. mathrand.Shuffle(j-i, func(a, b int) {
  1695. accounts[i+a], accounts[i+b] = accounts[i+b], accounts[i+a]
  1696. })
  1697. }
  1698. }
  1699. i = j
  1700. }
  1701. }
  1702. // sameAccountGroup 判断两个 Account 是否属于同一排序组(Priority + LastUsedAt)
  1703. func sameAccountGroup(a, b *Account) bool {
  1704. if a.Priority != b.Priority {
  1705. return false
  1706. }
  1707. return sameLastUsedAt(a.LastUsedAt, b.LastUsedAt)
  1708. }
  1709. // sameLastUsedAt 判断两个 LastUsedAt 是否相同(精度到秒)
  1710. func sameLastUsedAt(a, b *time.Time) bool {
  1711. switch {
  1712. case a == nil && b == nil:
  1713. return true
  1714. case a == nil || b == nil:
  1715. return false
  1716. default:
  1717. return a.Unix() == b.Unix()
  1718. }
  1719. }
  1720. // sortCandidatesForFallback 根据配置选择排序策略
  1721. // mode: "last_used"(按最后使用时间) 或 "random"(随机)
  1722. func (s *GatewayService) sortCandidatesForFallback(accounts []*Account, preferOAuth bool, mode string) {
  1723. if mode == "random" {
  1724. // 先按优先级排序,然后在同优先级内随机打乱
  1725. sortAccountsByPriorityOnly(accounts, preferOAuth)
  1726. shuffleWithinPriority(accounts)
  1727. } else {
  1728. // 默认按最后使用时间排序
  1729. sortAccountsByPriorityAndLastUsed(accounts, preferOAuth)
  1730. }
  1731. }
  1732. // sortAccountsByPriorityOnly 仅按优先级排序
  1733. func sortAccountsByPriorityOnly(accounts []*Account, preferOAuth bool) {
  1734. sort.SliceStable(accounts, func(i, j int) bool {
  1735. a, b := accounts[i], accounts[j]
  1736. if a.Priority != b.Priority {
  1737. return a.Priority < b.Priority
  1738. }
  1739. if preferOAuth && a.Type != b.Type {
  1740. return a.Type == AccountTypeOAuth
  1741. }
  1742. return false
  1743. })
  1744. }
  1745. // shuffleWithinPriority 在同优先级内随机打乱顺序
  1746. func shuffleWithinPriority(accounts []*Account) {
  1747. if len(accounts) <= 1 {
  1748. return
  1749. }
  1750. r := mathrand.New(mathrand.NewSource(time.Now().UnixNano()))
  1751. start := 0
  1752. for start < len(accounts) {
  1753. priority := accounts[start].Priority
  1754. end := start + 1
  1755. for end < len(accounts) && accounts[end].Priority == priority {
  1756. end++
  1757. }
  1758. // 对 [start, end) 范围内的账户随机打乱
  1759. if end-start > 1 {
  1760. r.Shuffle(end-start, func(i, j int) {
  1761. accounts[start+i], accounts[start+j] = accounts[start+j], accounts[start+i]
  1762. })
  1763. }
  1764. start = end
  1765. }
  1766. }
  1767. // selectAccountForModelWithPlatform 选择单平台账户(完全隔离)
  1768. func (s *GatewayService) selectAccountForModelWithPlatform(ctx context.Context, groupID *int64, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, platform string) (*Account, error) {
  1769. preferOAuth := platform == PlatformGemini
  1770. routingAccountIDs := s.routingAccountIDsForRequest(ctx, groupID, requestedModel, platform)
  1771. // require_privacy_set: 获取分组信息
  1772. var schedGroup *Group
  1773. if groupID != nil && s.groupRepo != nil {
  1774. schedGroup, _ = s.groupRepo.GetByID(ctx, *groupID)
  1775. }
  1776. var accounts []Account
  1777. accountsLoaded := false
  1778. // ============ Model Routing (legacy path): apply before sticky session ============
  1779. // When load-awareness is disabled (e.g. concurrency service not configured), we still honor model routing
  1780. // so switching model can switch upstream account within the same sticky session.
  1781. if len(routingAccountIDs) > 0 {
  1782. if s.debugModelRoutingEnabled() {
  1783. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] legacy routed begin: group_id=%v model=%s platform=%s session=%s routed_ids=%v",
  1784. derefGroupID(groupID), requestedModel, platform, shortSessionHash(sessionHash), routingAccountIDs)
  1785. }
  1786. // 1) Sticky session only applies if the bound account is within the routing set.
  1787. if sessionHash != "" && s.cache != nil {
  1788. accountID, err := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  1789. if err == nil && accountID > 0 && containsInt64(routingAccountIDs, accountID) {
  1790. if _, excluded := excludedIDs[accountID]; !excluded {
  1791. account, err := s.getSchedulableAccount(ctx, accountID)
  1792. // 检查账号分组归属和平台匹配(确保粘性会话不会跨分组或跨平台)
  1793. if err == nil {
  1794. clearSticky := shouldClearStickySession(account, requestedModel)
  1795. if clearSticky {
  1796. _ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  1797. }
  1798. if !clearSticky && s.isGatewayAccountProfitEligible(ctx, account) && s.isAccountInGroup(account, groupID) && account.Platform == platform && (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) && s.isAccountSchedulableForQuota(account) && s.isAccountSchedulableForWindowCost(ctx, account, true) && s.isAccountSchedulableForRPM(ctx, account, true) && !s.isStickyAccountUpstreamRestricted(ctx, groupID, account, requestedModel) {
  1799. if s.debugModelRoutingEnabled() {
  1800. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] legacy routed sticky hit: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), accountID)
  1801. }
  1802. return account, nil
  1803. }
  1804. }
  1805. }
  1806. }
  1807. }
  1808. // 2) Select an account from the routed candidates.
  1809. forcePlatform, hasForcePlatform := ctx.Value(ctxkey.ForcePlatform).(string)
  1810. if hasForcePlatform && forcePlatform == "" {
  1811. hasForcePlatform = false
  1812. }
  1813. var err error
  1814. accounts, _, err = s.listSchedulableAccounts(ctx, groupID, platform, hasForcePlatform)
  1815. if err != nil {
  1816. return nil, fmt.Errorf("query accounts failed: %w", err)
  1817. }
  1818. accountsLoaded = true
  1819. // 提前预取窗口费用+RPM 计数,确保 routing 段内的调度检查调用能命中缓存
  1820. ctx = s.withWindowCostPrefetch(ctx, accounts)
  1821. ctx = s.withRPMPrefetch(ctx, accounts)
  1822. routingSet := make(map[int64]struct{}, len(routingAccountIDs))
  1823. for _, id := range routingAccountIDs {
  1824. if id > 0 {
  1825. routingSet[id] = struct{}{}
  1826. }
  1827. }
  1828. var selected *Account
  1829. for i := range accounts {
  1830. acc := &accounts[i]
  1831. if _, ok := routingSet[acc.ID]; !ok {
  1832. continue
  1833. }
  1834. if _, excluded := excludedIDs[acc.ID]; excluded {
  1835. continue
  1836. }
  1837. // Scheduler snapshots can be temporarily stale; re-check schedulability here to
  1838. // avoid selecting accounts that were recently rate-limited/overloaded.
  1839. if !s.isAccountSchedulableForSelection(acc) {
  1840. continue
  1841. }
  1842. if !s.isGatewayAccountProfitEligible(ctx, acc) {
  1843. continue
  1844. }
  1845. // require_privacy_set: 跳过 privacy 未设置的账号并标记异常
  1846. if schedGroup != nil && schedGroup.RequirePrivacySet && !acc.IsPrivacySet() {
  1847. _ = s.accountRepo.SetError(ctx, acc.ID,
  1848. fmt.Sprintf("Privacy not set, required by group [%s]", schedGroup.Name))
  1849. continue
  1850. }
  1851. if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, acc, requestedModel) {
  1852. continue
  1853. }
  1854. if !s.isAccountSchedulableForModelSelection(ctx, acc, requestedModel) {
  1855. continue
  1856. }
  1857. if !s.isAccountSchedulableForQuota(acc) {
  1858. continue
  1859. }
  1860. if !s.isAccountSchedulableForWindowCost(ctx, acc, false) {
  1861. continue
  1862. }
  1863. if !s.isAccountSchedulableForRPM(ctx, acc, false) {
  1864. continue
  1865. }
  1866. if selected == nil {
  1867. selected = acc
  1868. continue
  1869. }
  1870. if acc.Priority < selected.Priority {
  1871. selected = acc
  1872. } else if acc.Priority == selected.Priority {
  1873. switch {
  1874. case acc.LastUsedAt == nil && selected.LastUsedAt != nil:
  1875. selected = acc
  1876. case acc.LastUsedAt != nil && selected.LastUsedAt == nil:
  1877. // keep selected (never used is preferred)
  1878. case acc.LastUsedAt == nil && selected.LastUsedAt == nil:
  1879. if preferOAuth && acc.Type != selected.Type && acc.Type == AccountTypeOAuth {
  1880. selected = acc
  1881. }
  1882. default:
  1883. if acc.LastUsedAt.Before(*selected.LastUsedAt) {
  1884. selected = acc
  1885. }
  1886. }
  1887. }
  1888. }
  1889. if selected != nil {
  1890. if sessionHash != "" && s.cache != nil {
  1891. if err := s.bindGatewayStickySessionDuringSelection(ctx, groupID, sessionHash, selected.ID); err != nil {
  1892. logger.LegacyPrintf("service.gateway", "set session account failed: session=%s account_id=%d err=%v", sessionHash, selected.ID, err)
  1893. }
  1894. }
  1895. if s.debugModelRoutingEnabled() {
  1896. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] legacy routed select: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), selected.ID)
  1897. }
  1898. return selected, nil
  1899. }
  1900. logger.LegacyPrintf("service.gateway", "[ModelRouting] No routed accounts available for model=%s, falling back to normal selection", requestedModel)
  1901. }
  1902. // 1. 查询粘性会话
  1903. if sessionHash != "" && s.cache != nil {
  1904. accountID, err := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  1905. if err == nil && accountID > 0 {
  1906. if _, excluded := excludedIDs[accountID]; !excluded {
  1907. account, err := s.getSchedulableAccount(ctx, accountID)
  1908. // 检查账号分组归属和平台匹配(确保粘性会话不会跨分组或跨平台)
  1909. if err == nil {
  1910. clearSticky := shouldClearStickySession(account, requestedModel)
  1911. if clearSticky {
  1912. _ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  1913. }
  1914. if !clearSticky && s.isGatewayAccountProfitEligible(ctx, account) && s.isAccountInGroup(account, groupID) && account.Platform == platform && (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) && s.isAccountSchedulableForQuota(account) && s.isAccountSchedulableForWindowCost(ctx, account, true) && s.isAccountSchedulableForRPM(ctx, account, true) {
  1915. return account, nil
  1916. }
  1917. }
  1918. }
  1919. }
  1920. }
  1921. // 2. 获取可调度账号列表(单平台)
  1922. if !accountsLoaded {
  1923. forcePlatform, hasForcePlatform := ctx.Value(ctxkey.ForcePlatform).(string)
  1924. if hasForcePlatform && forcePlatform == "" {
  1925. hasForcePlatform = false
  1926. }
  1927. var err error
  1928. accounts, _, err = s.listSchedulableAccounts(ctx, groupID, platform, hasForcePlatform)
  1929. if err != nil {
  1930. return nil, fmt.Errorf("query accounts failed: %w", err)
  1931. }
  1932. }
  1933. // 批量预取窗口费用+RPM 计数,避免逐个账号查询(N+1)
  1934. ctx = s.withWindowCostPrefetch(ctx, accounts)
  1935. ctx = s.withRPMPrefetch(ctx, accounts)
  1936. // 3. 按优先级+最久未用选择(考虑模型支持)
  1937. // needsUpstreamCheck 仅在主选择循环中使用;粘性会话命中时跳过此检查,
  1938. // 因为粘性会话优先保持连接一致性,且 upstream 计费基准极少使用。
  1939. needsUpstreamCheck := s.needsUpstreamChannelRestrictionCheck(ctx, groupID)
  1940. var selected *Account
  1941. for i := range accounts {
  1942. acc := &accounts[i]
  1943. if _, excluded := excludedIDs[acc.ID]; excluded {
  1944. continue
  1945. }
  1946. // Scheduler snapshots can be temporarily stale; re-check schedulability here to
  1947. // avoid selecting accounts that were recently rate-limited/overloaded.
  1948. if !s.isAccountSchedulableForSelection(acc) {
  1949. continue
  1950. }
  1951. if !s.isGatewayAccountProfitEligible(ctx, acc) {
  1952. continue
  1953. }
  1954. // require_privacy_set: 跳过 privacy 未设置的账号并标记异常
  1955. if schedGroup != nil && schedGroup.RequirePrivacySet && !acc.IsPrivacySet() {
  1956. _ = s.accountRepo.SetError(ctx, acc.ID,
  1957. fmt.Sprintf("Privacy not set, required by group [%s]", schedGroup.Name))
  1958. continue
  1959. }
  1960. if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, acc, requestedModel) {
  1961. continue
  1962. }
  1963. if needsUpstreamCheck && s.isUpstreamModelRestrictedByChannel(ctx, *groupID, acc, requestedModel) {
  1964. continue
  1965. }
  1966. if !s.isAccountSchedulableForModelSelection(ctx, acc, requestedModel) {
  1967. continue
  1968. }
  1969. if !s.isAccountSchedulableForQuota(acc) {
  1970. continue
  1971. }
  1972. if !s.isAccountSchedulableForWindowCost(ctx, acc, false) {
  1973. continue
  1974. }
  1975. if !s.isAccountSchedulableForRPM(ctx, acc, false) {
  1976. continue
  1977. }
  1978. if selected == nil {
  1979. selected = acc
  1980. continue
  1981. }
  1982. if acc.Priority < selected.Priority {
  1983. selected = acc
  1984. } else if acc.Priority == selected.Priority {
  1985. switch {
  1986. case acc.LastUsedAt == nil && selected.LastUsedAt != nil:
  1987. selected = acc
  1988. case acc.LastUsedAt != nil && selected.LastUsedAt == nil:
  1989. // keep selected (never used is preferred)
  1990. case acc.LastUsedAt == nil && selected.LastUsedAt == nil:
  1991. if preferOAuth && acc.Type != selected.Type && acc.Type == AccountTypeOAuth {
  1992. selected = acc
  1993. }
  1994. default:
  1995. if acc.LastUsedAt.Before(*selected.LastUsedAt) {
  1996. selected = acc
  1997. }
  1998. }
  1999. }
  2000. }
  2001. if selected == nil {
  2002. stats := s.logDetailedSelectionFailure(ctx, groupID, sessionHash, requestedModel, platform, accounts, excludedIDs, false)
  2003. if requestedModel != "" {
  2004. return nil, fmt.Errorf("%w supporting model: %s (%s)", ErrNoAvailableAccounts, requestedModel, summarizeSelectionFailureStats(stats))
  2005. }
  2006. return nil, ErrNoAvailableAccounts
  2007. }
  2008. // 4. 建立粘性绑定
  2009. if sessionHash != "" && s.cache != nil {
  2010. if err := s.bindGatewayStickySessionDuringSelection(ctx, groupID, sessionHash, selected.ID); err != nil {
  2011. logger.LegacyPrintf("service.gateway", "set session account failed: session=%s account_id=%d err=%v", sessionHash, selected.ID, err)
  2012. }
  2013. }
  2014. return selected, nil
  2015. }
  2016. // selectAccountWithMixedScheduling 选择账户(支持混合调度)
  2017. // 查询原生平台账户 + 启用 mixed_scheduling 的 antigravity 账户
  2018. func (s *GatewayService) selectAccountWithMixedScheduling(ctx context.Context, groupID *int64, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, nativePlatform string) (*Account, error) {
  2019. preferOAuth := nativePlatform == PlatformGemini
  2020. routingAccountIDs := s.routingAccountIDsForRequest(ctx, groupID, requestedModel, nativePlatform)
  2021. // require_privacy_set: 获取分组信息
  2022. var schedGroup *Group
  2023. if groupID != nil && s.groupRepo != nil {
  2024. schedGroup, _ = s.groupRepo.GetByID(ctx, *groupID)
  2025. }
  2026. var accounts []Account
  2027. accountsLoaded := false
  2028. // ============ Model Routing (legacy path): apply before sticky session ============
  2029. if len(routingAccountIDs) > 0 {
  2030. if s.debugModelRoutingEnabled() {
  2031. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] legacy mixed routed begin: group_id=%v model=%s platform=%s session=%s routed_ids=%v",
  2032. derefGroupID(groupID), requestedModel, nativePlatform, shortSessionHash(sessionHash), routingAccountIDs)
  2033. }
  2034. // 1) Sticky session only applies if the bound account is within the routing set.
  2035. if sessionHash != "" && s.cache != nil {
  2036. accountID, err := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  2037. if err == nil && accountID > 0 && containsInt64(routingAccountIDs, accountID) {
  2038. if _, excluded := excludedIDs[accountID]; !excluded {
  2039. account, err := s.getSchedulableAccount(ctx, accountID)
  2040. // 检查账号分组归属和有效性:原生平台直接匹配,antigravity 需要启用混合调度
  2041. if err == nil {
  2042. clearSticky := shouldClearStickySession(account, requestedModel)
  2043. if clearSticky {
  2044. _ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  2045. }
  2046. if !clearSticky && s.isGatewayAccountProfitEligible(ctx, account) && s.isAccountInGroup(account, groupID) && (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) && s.isAccountSchedulableForQuota(account) && s.isAccountSchedulableForWindowCost(ctx, account, true) && s.isAccountSchedulableForRPM(ctx, account, true) {
  2047. if account.Platform == nativePlatform || (account.Platform == PlatformAntigravity && account.IsMixedSchedulingEnabled()) {
  2048. if s.debugModelRoutingEnabled() {
  2049. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] legacy mixed routed sticky hit: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), accountID)
  2050. }
  2051. return account, nil
  2052. }
  2053. }
  2054. }
  2055. }
  2056. }
  2057. }
  2058. // 2) Select an account from the routed candidates.
  2059. var err error
  2060. accounts, _, err = s.listSchedulableAccounts(ctx, groupID, nativePlatform, false)
  2061. if err != nil {
  2062. return nil, fmt.Errorf("query accounts failed: %w", err)
  2063. }
  2064. accountsLoaded = true
  2065. // 提前预取窗口费用+RPM 计数,确保 routing 段内的调度检查调用能命中缓存
  2066. ctx = s.withWindowCostPrefetch(ctx, accounts)
  2067. ctx = s.withRPMPrefetch(ctx, accounts)
  2068. routingSet := make(map[int64]struct{}, len(routingAccountIDs))
  2069. for _, id := range routingAccountIDs {
  2070. if id > 0 {
  2071. routingSet[id] = struct{}{}
  2072. }
  2073. }
  2074. var selected *Account
  2075. for i := range accounts {
  2076. acc := &accounts[i]
  2077. if _, ok := routingSet[acc.ID]; !ok {
  2078. continue
  2079. }
  2080. if _, excluded := excludedIDs[acc.ID]; excluded {
  2081. continue
  2082. }
  2083. // Scheduler snapshots can be temporarily stale; re-check schedulability here to
  2084. // avoid selecting accounts that were recently rate-limited/overloaded.
  2085. if !s.isAccountSchedulableForSelection(acc) {
  2086. continue
  2087. }
  2088. if !s.isGatewayAccountProfitEligible(ctx, acc) {
  2089. continue
  2090. }
  2091. // require_privacy_set: 跳过 privacy 未设置的账号并标记异常
  2092. if schedGroup != nil && schedGroup.RequirePrivacySet && !acc.IsPrivacySet() {
  2093. _ = s.accountRepo.SetError(ctx, acc.ID,
  2094. fmt.Sprintf("Privacy not set, required by group [%s]", schedGroup.Name))
  2095. continue
  2096. }
  2097. // 过滤:原生平台直接通过,antigravity 需要启用混合调度
  2098. if acc.Platform == PlatformAntigravity && !acc.IsMixedSchedulingEnabled() {
  2099. continue
  2100. }
  2101. if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, acc, requestedModel) {
  2102. continue
  2103. }
  2104. if !s.isAccountSchedulableForModelSelection(ctx, acc, requestedModel) {
  2105. continue
  2106. }
  2107. if !s.isAccountSchedulableForQuota(acc) {
  2108. continue
  2109. }
  2110. if !s.isAccountSchedulableForWindowCost(ctx, acc, false) {
  2111. continue
  2112. }
  2113. if !s.isAccountSchedulableForRPM(ctx, acc, false) {
  2114. continue
  2115. }
  2116. if selected == nil {
  2117. selected = acc
  2118. continue
  2119. }
  2120. if acc.Priority < selected.Priority {
  2121. selected = acc
  2122. } else if acc.Priority == selected.Priority {
  2123. switch {
  2124. case acc.LastUsedAt == nil && selected.LastUsedAt != nil:
  2125. selected = acc
  2126. case acc.LastUsedAt != nil && selected.LastUsedAt == nil:
  2127. // keep selected (never used is preferred)
  2128. case acc.LastUsedAt == nil && selected.LastUsedAt == nil:
  2129. if preferOAuth && acc.Platform == PlatformGemini && selected.Platform == PlatformGemini && acc.Type != selected.Type && acc.Type == AccountTypeOAuth {
  2130. selected = acc
  2131. }
  2132. default:
  2133. if acc.LastUsedAt.Before(*selected.LastUsedAt) {
  2134. selected = acc
  2135. }
  2136. }
  2137. }
  2138. }
  2139. if selected != nil {
  2140. if sessionHash != "" && s.cache != nil {
  2141. if err := s.bindGatewayStickySessionDuringSelection(ctx, groupID, sessionHash, selected.ID); err != nil {
  2142. logger.LegacyPrintf("service.gateway", "set session account failed: session=%s account_id=%d err=%v", sessionHash, selected.ID, err)
  2143. }
  2144. }
  2145. if s.debugModelRoutingEnabled() {
  2146. logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] legacy mixed routed select: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), selected.ID)
  2147. }
  2148. return selected, nil
  2149. }
  2150. logger.LegacyPrintf("service.gateway", "[ModelRouting] No routed accounts available for model=%s, falling back to normal selection", requestedModel)
  2151. }
  2152. // 1. 查询粘性会话
  2153. if sessionHash != "" && s.cache != nil {
  2154. accountID, err := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  2155. if err == nil && accountID > 0 {
  2156. if _, excluded := excludedIDs[accountID]; !excluded {
  2157. account, err := s.getSchedulableAccount(ctx, accountID)
  2158. // 检查账号分组归属和有效性:原生平台直接匹配,antigravity 需要启用混合调度
  2159. if err == nil {
  2160. clearSticky := shouldClearStickySession(account, requestedModel)
  2161. if clearSticky {
  2162. _ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
  2163. }
  2164. if !clearSticky && s.isGatewayAccountProfitEligible(ctx, account) && s.isAccountInGroup(account, groupID) && (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) && s.isAccountSchedulableForQuota(account) && s.isAccountSchedulableForWindowCost(ctx, account, true) && s.isAccountSchedulableForRPM(ctx, account, true) && !s.isStickyAccountUpstreamRestricted(ctx, groupID, account, requestedModel) {
  2165. if account.Platform == nativePlatform || (account.Platform == PlatformAntigravity && account.IsMixedSchedulingEnabled()) {
  2166. return account, nil
  2167. }
  2168. }
  2169. }
  2170. }
  2171. }
  2172. }
  2173. // 2. 获取可调度账号列表
  2174. if !accountsLoaded {
  2175. var err error
  2176. accounts, _, err = s.listSchedulableAccounts(ctx, groupID, nativePlatform, false)
  2177. if err != nil {
  2178. return nil, fmt.Errorf("query accounts failed: %w", err)
  2179. }
  2180. }
  2181. // 批量预取窗口费用+RPM 计数,避免逐个账号查询(N+1)
  2182. ctx = s.withWindowCostPrefetch(ctx, accounts)
  2183. ctx = s.withRPMPrefetch(ctx, accounts)
  2184. // 3. 按优先级+最久未用选择(考虑模型支持和混合调度)
  2185. // needsUpstreamCheck 仅在主选择循环中使用;粘性会话命中时跳过此检查。
  2186. needsUpstreamCheck := s.needsUpstreamChannelRestrictionCheck(ctx, groupID)
  2187. var selected *Account
  2188. for i := range accounts {
  2189. acc := &accounts[i]
  2190. if _, excluded := excludedIDs[acc.ID]; excluded {
  2191. continue
  2192. }
  2193. // Scheduler snapshots can be temporarily stale; re-check schedulability here to
  2194. // avoid selecting accounts that were recently rate-limited/overloaded.
  2195. if !s.isAccountSchedulableForSelection(acc) {
  2196. continue
  2197. }
  2198. if !s.isGatewayAccountProfitEligible(ctx, acc) {
  2199. continue
  2200. }
  2201. // require_privacy_set: 跳过 privacy 未设置的账号并标记异常
  2202. if schedGroup != nil && schedGroup.RequirePrivacySet && !acc.IsPrivacySet() {
  2203. _ = s.accountRepo.SetError(ctx, acc.ID,
  2204. fmt.Sprintf("Privacy not set, required by group [%s]", schedGroup.Name))
  2205. continue
  2206. }
  2207. // 过滤:原生平台直接通过,antigravity 需要启用混合调度
  2208. if acc.Platform == PlatformAntigravity && !acc.IsMixedSchedulingEnabled() {
  2209. continue
  2210. }
  2211. if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, acc, requestedModel) {
  2212. continue
  2213. }
  2214. if needsUpstreamCheck && s.isUpstreamModelRestrictedByChannel(ctx, *groupID, acc, requestedModel) {
  2215. continue
  2216. }
  2217. if !s.isAccountSchedulableForModelSelection(ctx, acc, requestedModel) {
  2218. continue
  2219. }
  2220. if !s.isAccountSchedulableForQuota(acc) {
  2221. continue
  2222. }
  2223. if !s.isAccountSchedulableForWindowCost(ctx, acc, false) {
  2224. continue
  2225. }
  2226. if !s.isAccountSchedulableForRPM(ctx, acc, false) {
  2227. continue
  2228. }
  2229. if selected == nil {
  2230. selected = acc
  2231. continue
  2232. }
  2233. if acc.Priority < selected.Priority {
  2234. selected = acc
  2235. } else if acc.Priority == selected.Priority {
  2236. switch {
  2237. case acc.LastUsedAt == nil && selected.LastUsedAt != nil:
  2238. selected = acc
  2239. case acc.LastUsedAt != nil && selected.LastUsedAt == nil:
  2240. // keep selected (never used is preferred)
  2241. case acc.LastUsedAt == nil && selected.LastUsedAt == nil:
  2242. if preferOAuth && acc.Platform == PlatformGemini && selected.Platform == PlatformGemini && acc.Type != selected.Type && acc.Type == AccountTypeOAuth {
  2243. selected = acc
  2244. }
  2245. default:
  2246. if acc.LastUsedAt.Before(*selected.LastUsedAt) {
  2247. selected = acc
  2248. }
  2249. }
  2250. }
  2251. }
  2252. if selected == nil {
  2253. stats := s.logDetailedSelectionFailure(ctx, groupID, sessionHash, requestedModel, nativePlatform, accounts, excludedIDs, true)
  2254. if requestedModel != "" {
  2255. return nil, fmt.Errorf("%w supporting model: %s (%s)", ErrNoAvailableAccounts, requestedModel, summarizeSelectionFailureStats(stats))
  2256. }
  2257. return nil, ErrNoAvailableAccounts
  2258. }
  2259. // 4. 建立粘性绑定
  2260. if sessionHash != "" && s.cache != nil {
  2261. if err := s.bindGatewayStickySessionDuringSelection(ctx, groupID, sessionHash, selected.ID); err != nil {
  2262. logger.LegacyPrintf("service.gateway", "set session account failed: session=%s account_id=%d err=%v", sessionHash, selected.ID, err)
  2263. }
  2264. }
  2265. return selected, nil
  2266. }
  2267. type selectionFailureStats struct {
  2268. Total int
  2269. Eligible int
  2270. Excluded int
  2271. Unschedulable int
  2272. PlatformFiltered int
  2273. ModelUnsupported int
  2274. ModelRateLimited int
  2275. ProfitThreshold int
  2276. ProfitInvalidRate int
  2277. SamplePlatformIDs []int64
  2278. SampleMappingIDs []int64
  2279. SampleRateLimitIDs []string
  2280. }
  2281. type selectionFailureDiagnosis struct {
  2282. Category string
  2283. Detail string
  2284. }
  2285. func (s *GatewayService) logDetailedSelectionFailure(
  2286. ctx context.Context,
  2287. groupID *int64,
  2288. sessionHash string,
  2289. requestedModel string,
  2290. platform string,
  2291. accounts []Account,
  2292. excludedIDs map[int64]struct{},
  2293. allowMixedScheduling bool,
  2294. ) selectionFailureStats {
  2295. stats := s.collectSelectionFailureStats(ctx, accounts, requestedModel, platform, excludedIDs, allowMixedScheduling)
  2296. logger.LegacyPrintf(
  2297. "service.gateway",
  2298. "[SelectAccountDetailed] group_id=%v model=%s platform=%s session=%s total=%d eligible=%d excluded=%d unschedulable=%d platform_filtered=%d model_unsupported=%d model_rate_limited=%d profit_threshold=%d profit_invalid_account_rate=%d sample_platform_filtered=%v sample_model_unsupported=%v sample_model_rate_limited=%v",
  2299. derefGroupID(groupID),
  2300. requestedModel,
  2301. platform,
  2302. shortSessionHash(sessionHash),
  2303. stats.Total,
  2304. stats.Eligible,
  2305. stats.Excluded,
  2306. stats.Unschedulable,
  2307. stats.PlatformFiltered,
  2308. stats.ModelUnsupported,
  2309. stats.ModelRateLimited,
  2310. stats.ProfitThreshold,
  2311. stats.ProfitInvalidRate,
  2312. stats.SamplePlatformIDs,
  2313. stats.SampleMappingIDs,
  2314. stats.SampleRateLimitIDs,
  2315. )
  2316. return stats
  2317. }
  2318. func (s *GatewayService) collectSelectionFailureStats(
  2319. ctx context.Context,
  2320. accounts []Account,
  2321. requestedModel string,
  2322. platform string,
  2323. excludedIDs map[int64]struct{},
  2324. allowMixedScheduling bool,
  2325. ) selectionFailureStats {
  2326. stats := selectionFailureStats{
  2327. Total: len(accounts),
  2328. }
  2329. for i := range accounts {
  2330. acc := &accounts[i]
  2331. diagnosis := s.diagnoseSelectionFailure(ctx, acc, requestedModel, platform, excludedIDs, allowMixedScheduling)
  2332. switch diagnosis.Category {
  2333. case "excluded":
  2334. stats.Excluded++
  2335. case "unschedulable":
  2336. stats.Unschedulable++
  2337. case "platform_filtered":
  2338. stats.PlatformFiltered++
  2339. stats.SamplePlatformIDs = appendSelectionFailureSampleID(stats.SamplePlatformIDs, acc.ID)
  2340. case "model_unsupported":
  2341. stats.ModelUnsupported++
  2342. stats.SampleMappingIDs = appendSelectionFailureSampleID(stats.SampleMappingIDs, acc.ID)
  2343. case "model_rate_limited":
  2344. stats.ModelRateLimited++
  2345. remaining := acc.GetRateLimitRemainingTimeWithContext(ctx, requestedModel).Truncate(time.Second)
  2346. stats.SampleRateLimitIDs = appendSelectionFailureRateSample(stats.SampleRateLimitIDs, acc.ID, remaining)
  2347. case openAIProfitFilterReasonThreshold:
  2348. stats.ProfitThreshold++
  2349. case openAIProfitFilterReasonInvalidAccountRate:
  2350. stats.ProfitInvalidRate++
  2351. default:
  2352. stats.Eligible++
  2353. }
  2354. }
  2355. return stats
  2356. }
  2357. func (s *GatewayService) diagnoseSelectionFailure(
  2358. ctx context.Context,
  2359. acc *Account,
  2360. requestedModel string,
  2361. platform string,
  2362. excludedIDs map[int64]struct{},
  2363. allowMixedScheduling bool,
  2364. ) selectionFailureDiagnosis {
  2365. if acc == nil {
  2366. return selectionFailureDiagnosis{Category: "unschedulable", Detail: "account_nil"}
  2367. }
  2368. if _, excluded := excludedIDs[acc.ID]; excluded {
  2369. return selectionFailureDiagnosis{Category: "excluded"}
  2370. }
  2371. if !s.isAccountSchedulableForSelection(acc) {
  2372. return selectionFailureDiagnosis{Category: "unschedulable", Detail: "generic_unschedulable"}
  2373. }
  2374. if isPlatformFilteredForSelection(acc, platform, allowMixedScheduling) {
  2375. return selectionFailureDiagnosis{
  2376. Category: "platform_filtered",
  2377. Detail: fmt.Sprintf("account_platform=%s requested_platform=%s", acc.Platform, strings.TrimSpace(platform)),
  2378. }
  2379. }
  2380. if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, acc, requestedModel) {
  2381. return selectionFailureDiagnosis{
  2382. Category: "model_unsupported",
  2383. Detail: fmt.Sprintf("model=%s", requestedModel),
  2384. }
  2385. }
  2386. if !s.isAccountSchedulableForModelSelection(ctx, acc, requestedModel) {
  2387. remaining := acc.GetRateLimitRemainingTimeWithContext(ctx, requestedModel).Truncate(time.Second)
  2388. return selectionFailureDiagnosis{
  2389. Category: "model_rate_limited",
  2390. Detail: fmt.Sprintf("remaining=%s", remaining),
  2391. }
  2392. }
  2393. if vetoed, reason := openAIProfitControlVetoReason(ctx, acc); vetoed {
  2394. return selectionFailureDiagnosis{Category: reason}
  2395. }
  2396. return selectionFailureDiagnosis{Category: "eligible"}
  2397. }
  2398. func isPlatformFilteredForSelection(acc *Account, platform string, allowMixedScheduling bool) bool {
  2399. if acc == nil {
  2400. return true
  2401. }
  2402. if allowMixedScheduling {
  2403. if acc.Platform == PlatformAntigravity {
  2404. return !acc.IsMixedSchedulingEnabled()
  2405. }
  2406. return acc.Platform != platform
  2407. }
  2408. if strings.TrimSpace(platform) == "" {
  2409. return false
  2410. }
  2411. return acc.Platform != platform
  2412. }
  2413. func appendSelectionFailureSampleID(samples []int64, id int64) []int64 {
  2414. const limit = 5
  2415. if len(samples) >= limit {
  2416. return samples
  2417. }
  2418. return append(samples, id)
  2419. }
  2420. func appendSelectionFailureRateSample(samples []string, accountID int64, remaining time.Duration) []string {
  2421. const limit = 5
  2422. if len(samples) >= limit {
  2423. return samples
  2424. }
  2425. return append(samples, fmt.Sprintf("%d(%s)", accountID, remaining))
  2426. }
  2427. func summarizeSelectionFailureStats(stats selectionFailureStats) string {
  2428. return fmt.Sprintf(
  2429. "total=%d eligible=%d excluded=%d unschedulable=%d platform_filtered=%d model_unsupported=%d model_rate_limited=%d profit_threshold=%d profit_invalid_account_rate=%d",
  2430. stats.Total,
  2431. stats.Eligible,
  2432. stats.Excluded,
  2433. stats.Unschedulable,
  2434. stats.PlatformFiltered,
  2435. stats.ModelUnsupported,
  2436. stats.ModelRateLimited,
  2437. stats.ProfitThreshold,
  2438. stats.ProfitInvalidRate,
  2439. )
  2440. }
  2441. // isModelSupportedByAccountWithContext 根据账户平台检查模型支持(带 context)
  2442. // 对于 Antigravity 平台,会先获取映射后的最终模型名(包括 thinking 后缀)再检查支持
  2443. func (s *GatewayService) isModelSupportedByAccountWithContext(ctx context.Context, account *Account, requestedModel string) bool {
  2444. if source, ok := CompositeRouteSourceFromContext(ctx); ok && source == CompositeRouteSourceAccount {
  2445. if publicModel, modelOK := RequestedPublicModelFromContext(ctx); modelOK && !explicitModelMappingClaims(*account, publicModel) {
  2446. return false
  2447. }
  2448. }
  2449. if account.Platform == PlatformAntigravity {
  2450. if strings.TrimSpace(requestedModel) == "" {
  2451. return true
  2452. }
  2453. // 使用与转发阶段一致的映射逻辑:自定义映射优先 → 默认映射兜底
  2454. mapped := mapAntigravityModel(account, requestedModel)
  2455. if mapped == "" {
  2456. return false
  2457. }
  2458. // 应用 thinking 后缀后检查最终模型是否在账号映射中
  2459. if enabled, ok := ThinkingEnabledFromContext(ctx); ok {
  2460. finalModel := applyThinkingModelSuffix(mapped, enabled)
  2461. if finalModel == mapped {
  2462. return true // thinking 后缀未改变模型名,映射已通过
  2463. }
  2464. return account.IsModelSupported(finalModel)
  2465. }
  2466. return true
  2467. }
  2468. return s.isModelSupportedByAccount(account, requestedModel)
  2469. }
  2470. // isModelSupportedByAccount 根据账户平台检查模型支持(无 context,用于非 Antigravity 平台)
  2471. func (s *GatewayService) isModelSupportedByAccount(account *Account, requestedModel string) bool {
  2472. if account.Platform == PlatformAntigravity {
  2473. if strings.TrimSpace(requestedModel) == "" {
  2474. return true
  2475. }
  2476. return mapAntigravityModel(account, requestedModel) != ""
  2477. }
  2478. if account.IsBedrock() {
  2479. _, ok := ResolveBedrockModelID(account, requestedModel)
  2480. return ok
  2481. }
  2482. // OpenAI 透传模式:仅替换认证,允许所有模型
  2483. if account.Platform == PlatformOpenAI && account.IsOpenAIPassthroughEnabled() {
  2484. return true
  2485. }
  2486. // OAuth/SetupToken 账号使用 Anthropic 标准映射(短ID → 长ID)
  2487. if account.Platform == PlatformAnthropic && account.Type != AccountTypeAPIKey {
  2488. if account.Type == AccountTypeServiceAccount {
  2489. requestedModel = normalizeVertexAnthropicModelID(claude.NormalizeModelID(requestedModel))
  2490. } else {
  2491. requestedModel = claude.NormalizeModelID(requestedModel)
  2492. }
  2493. }
  2494. // 其他平台使用账户的模型支持检查
  2495. return account.IsModelSupported(requestedModel)
  2496. }