mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-11-01 08:28:34 +00:00
Merge pull request #7427 from xlplbo/fix/openai-legacy-scheduler-decision
fix(openai): 非高级调度补齐选号决策并按 previous_response 路由
This commit is contained in:
@@ -580,6 +580,8 @@ type AccountSelectionResult struct {
|
||||
Acquired bool
|
||||
ReleaseFunc func()
|
||||
WaitPlan *AccountWaitPlan // nil means no wait allowed
|
||||
// stickySessionHit 标记账号来自会话粘性绑定命中,供非高级调度路径回填决策标签。
|
||||
stickySessionHit bool
|
||||
// profitGate 携带本次选号真实生效的利润门(无门为 nil)。门安装在调度栈的
|
||||
// 局部 ctx 上,handler 必须经 ContextWithSelectionProfitGate 重放后才能在
|
||||
// 调度栈之外做抢槽后终检与准入后粘性绑定。
|
||||
|
||||
@@ -2222,6 +2222,74 @@ func (s *OpenAIGatewayService) loadOpenAIGroupRequiresPrivacySet(ctx context.Con
|
||||
return group != nil && group.RequirePrivacySet
|
||||
}
|
||||
|
||||
// applyLegacySelectionDecision 把非高级调度路径的选号结果回填到决策:记录选中账号,
|
||||
// 命中会话粘性时标记 session_hash 层,否则保持 load_balance。
|
||||
func applyLegacySelectionDecision(decision *OpenAIAccountScheduleDecision, selection *AccountSelectionResult) {
|
||||
if decision == nil || selection == nil || selection.Account == nil {
|
||||
return
|
||||
}
|
||||
decision.SelectedAccountID = selection.Account.ID
|
||||
decision.SelectedAccountType = selection.Account.Type
|
||||
if selection.stickySessionHit {
|
||||
decision.Layer = openAIAccountScheduleLayerSessionSticky
|
||||
decision.StickySessionHit = true
|
||||
}
|
||||
}
|
||||
|
||||
// selectLegacyAccountByPreviousResponse 在非高级调度路径按 previous_response_id 命中持有该响应的账号,
|
||||
// 先按请求模型做渠道限制检查,再复用高级调度器 previous_response 层的账号兼容校验
|
||||
// (分组、隐私、运行期封禁、代理隔离、上游模型渠道限制、传输与能力);未命中返回 (nil, false, nil)。
|
||||
func (s *OpenAIGatewayService) selectLegacyAccountByPreviousResponse(
|
||||
ctx context.Context,
|
||||
groupID *int64,
|
||||
previousResponseID string,
|
||||
sessionHash string,
|
||||
requestedModel string,
|
||||
excludedIDs map[int64]struct{},
|
||||
requiredTransport OpenAIUpstreamTransport,
|
||||
requiredCapability OpenAIEndpointCapability,
|
||||
requiredImageCapability OpenAIImagesCapability,
|
||||
requireCompact bool,
|
||||
platform string,
|
||||
) (*AccountSelectionResult, bool, error) {
|
||||
if strings.TrimSpace(previousResponseID) == "" || platform != PlatformOpenAI {
|
||||
return nil, false, nil
|
||||
}
|
||||
if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) {
|
||||
return nil, false, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
|
||||
}
|
||||
selection, err := s.selectAccountByPreviousResponseIDForCapability(ctx, groupID, previousResponseID, requestedModel, excludedIDs, requiredCapability, requireCompact)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
if selection == nil || selection.Account == nil {
|
||||
return nil, false, nil
|
||||
}
|
||||
account := selection.Account
|
||||
scheduler := &defaultOpenAIAccountScheduler{service: s, stats: newOpenAIAccountRuntimeStats()}
|
||||
compatible, _ := scheduler.isAccountRequestCompatibleReason(ctx, account, OpenAIAccountScheduleRequest{
|
||||
GroupID: groupID,
|
||||
Platform: platform,
|
||||
RequestedModel: requestedModel,
|
||||
RequiredTransport: requiredTransport,
|
||||
RequiredCapability: requiredCapability,
|
||||
RequiredImageCapability: requiredImageCapability,
|
||||
RequireCompact: requireCompact,
|
||||
ExcludedIDs: excludedIDs,
|
||||
RequirePrivacySet: s.openAIGroupRequiresPrivacySet(ctx, groupID),
|
||||
})
|
||||
if !s.openAIAccountMatchesSchedulingGroup(account, groupID) || !compatible || !scheduler.isAccountTransportCompatible(account, requiredTransport) {
|
||||
if selection.ReleaseFunc != nil {
|
||||
selection.ReleaseFunc()
|
||||
}
|
||||
return nil, false, nil
|
||||
}
|
||||
if sessionHash != "" {
|
||||
_ = s.bindOpenAIStickySessionDuringSelection(ctx, groupID, sessionHash, account.ID)
|
||||
}
|
||||
return selection, true, nil
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
|
||||
ctx context.Context,
|
||||
groupID *int64,
|
||||
@@ -2258,6 +2326,15 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
|
||||
scheduler := s.getOpenAIAccountScheduler(ctx)
|
||||
if scheduler == nil {
|
||||
decision.Layer = openAIAccountScheduleLayerLoadBalance
|
||||
if selection, hit, err := s.selectLegacyAccountByPreviousResponse(ctx, groupID, previousResponseID, sessionHash, requestedModel, excludedIDs, requiredTransport, requiredCapability, requiredImageCapability, requireCompact, platform); err != nil {
|
||||
return nil, decision, err
|
||||
} else if hit {
|
||||
decision.Layer = openAIAccountScheduleLayerPreviousResponse
|
||||
decision.StickyPreviousHit = true
|
||||
decision.SelectedAccountID = selection.Account.ID
|
||||
decision.SelectedAccountType = selection.Account.Type
|
||||
return selection, decision, nil
|
||||
}
|
||||
if guardianParentAccountID > 0 {
|
||||
if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) {
|
||||
return nil, decision, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
|
||||
@@ -2303,6 +2380,7 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
|
||||
return selection, decision, nil
|
||||
}
|
||||
if accountSupportsOpenAICapabilities(selection.Account, requiredCapability, requiredImageCapability) {
|
||||
applyLegacySelectionDecision(&decision, selection)
|
||||
return selection, decision, nil
|
||||
}
|
||||
if selection.ReleaseFunc != nil {
|
||||
@@ -2329,6 +2407,7 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
|
||||
}
|
||||
if s.isOpenAIAccountTransportCompatible(selection.Account, requiredTransport) &&
|
||||
accountSupportsOpenAICapabilities(selection.Account, requiredCapability, requiredImageCapability) {
|
||||
applyLegacySelectionDecision(&decision, selection)
|
||||
return selection, decision, nil
|
||||
}
|
||||
if selection.ReleaseFunc != nil {
|
||||
|
||||
@@ -882,25 +882,32 @@ func resolveOpenAIErrorSchedulingModel(billingModel, upstreamModel string) strin
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.Context, groupID *int64, platform string, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, requireCompact bool, stickyAccountID int64, requiredCapability OpenAIEndpointCapability, preferLowUpstreamRate bool) (*Account, error) {
|
||||
account, _, err := s.selectAccountForModelWithExclusionsStickyHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate)
|
||||
return account, err
|
||||
}
|
||||
|
||||
// selectAccountForModelWithExclusionsStickyHit 与 selectAccountForModelWithExclusions 相同,
|
||||
// 另返回账号是否来自粘性会话命中。
|
||||
func (s *OpenAIGatewayService) selectAccountForModelWithExclusionsStickyHit(ctx context.Context, groupID *int64, platform string, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, requireCompact bool, stickyAccountID int64, requiredCapability OpenAIEndpointCapability, preferLowUpstreamRate bool) (*Account, bool, error) {
|
||||
platform = NormalizeOpenAICompatiblePlatform(platform)
|
||||
if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) {
|
||||
slog.Warn("channel pricing restriction blocked request",
|
||||
"group_id", derefGroupID(groupID),
|
||||
"model", requestedModel)
|
||||
return nil, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
|
||||
return nil, false, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
|
||||
}
|
||||
|
||||
// 1. 尝试粘性会话命中
|
||||
// Try sticky session hit
|
||||
if account := s.tryStickySessionHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability); account != nil {
|
||||
return account, nil
|
||||
return account, true, nil
|
||||
}
|
||||
|
||||
// 2. 获取可调度的 OpenAI 账号
|
||||
// Get schedulable OpenAI accounts
|
||||
accounts, err := s.listSchedulableAccounts(ctx, groupID, platform)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query accounts failed: %w", err)
|
||||
return nil, false, fmt.Errorf("query accounts failed: %w", err)
|
||||
}
|
||||
|
||||
// 3. 按优先级 + LRU 选择最佳账号
|
||||
@@ -908,12 +915,12 @@ func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.C
|
||||
selected, compactBlocked, filterStats := s.selectBestAccount(ctx, groupID, platform, accounts, requestedModel, excludedIDs, requireCompact, requiredCapability, preferLowUpstreamRate)
|
||||
|
||||
if selected == nil {
|
||||
return nil, noAvailableOpenAISelectionError(requestedModel, compactBlocked, filterStats.summary(""))
|
||||
return nil, false, noAvailableOpenAISelectionError(requestedModel, compactBlocked, filterStats.summary(""))
|
||||
}
|
||||
|
||||
hydrated, err := s.hydrateSelectedAccount(ctx, selected)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
// 4. 设置粘性会话绑定(利润门下推迟到 handler 终检通过后再绑定,
|
||||
@@ -923,7 +930,7 @@ func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.C
|
||||
_ = s.setStickySessionAccountID(ctx, groupID, sessionHash, selected.ID, openaiStickySessionTTL)
|
||||
}
|
||||
|
||||
return hydrated, nil
|
||||
return hydrated, false, nil
|
||||
}
|
||||
|
||||
// tryStickySessionHit 尝试从粘性会话获取账号。
|
||||
@@ -1133,31 +1140,34 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex
|
||||
}
|
||||
}
|
||||
if s.concurrencyService == nil || !cfg.LoadBatchEnabled {
|
||||
account, err := s.selectAccountForModelWithExclusions(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate)
|
||||
account, stickyHit, err := s.selectAccountForModelWithExclusionsStickyHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result, err := s.tryAcquireAccountSlot(ctx, account.ID, account.Concurrency)
|
||||
if err == nil && result != nil && result.Acquired {
|
||||
return s.newAcquiredSelectionResult(ctx, account, result.ReleaseFunc)
|
||||
selection, selectErr := s.newAcquiredSelectionResult(ctx, account, result.ReleaseFunc)
|
||||
return markStickySessionHit(selection, stickyHit), selectErr
|
||||
}
|
||||
if stickyAccountID > 0 && stickyAccountID == account.ID && s.concurrencyService != nil {
|
||||
waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, account.ID)
|
||||
if waitingCount < cfg.StickySessionMaxWaiting {
|
||||
return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
|
||||
selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
|
||||
AccountID: account.ID,
|
||||
MaxConcurrency: account.Concurrency,
|
||||
Timeout: cfg.StickySessionWaitTimeout,
|
||||
MaxWaiting: cfg.StickySessionMaxWaiting,
|
||||
})
|
||||
return markStickySessionHit(selection, stickyHit), selectErr
|
||||
}
|
||||
}
|
||||
return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
|
||||
selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
|
||||
AccountID: account.ID,
|
||||
MaxConcurrency: account.Concurrency,
|
||||
Timeout: cfg.FallbackWaitTimeout,
|
||||
MaxWaiting: cfg.FallbackMaxWaiting,
|
||||
})
|
||||
return markStickySessionHit(selection, stickyHit), selectErr
|
||||
}
|
||||
|
||||
accounts, err := s.listSchedulableAccounts(ctx, groupID, platform)
|
||||
@@ -1211,17 +1221,18 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex
|
||||
return nil, selectErr
|
||||
}
|
||||
_ = s.refreshStickySessionTTL(ctx, groupID, sessionHash, openaiStickySessionTTL)
|
||||
return selection, nil
|
||||
return markStickySessionHit(selection, true), nil
|
||||
}
|
||||
|
||||
waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, accountID)
|
||||
if waitingCount < cfg.StickySessionMaxWaiting {
|
||||
return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
|
||||
selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
|
||||
AccountID: accountID,
|
||||
MaxConcurrency: account.Concurrency,
|
||||
Timeout: cfg.StickySessionWaitTimeout,
|
||||
MaxWaiting: cfg.StickySessionMaxWaiting,
|
||||
})
|
||||
return markStickySessionHit(selection, true), selectErr
|
||||
}
|
||||
stickySpillover = true
|
||||
}
|
||||
@@ -1730,6 +1741,14 @@ func (s *OpenAIGatewayService) newAcquiredSelectionResult(ctx context.Context, a
|
||||
return selection, err
|
||||
}
|
||||
|
||||
// markStickySessionHit 在选号结果上记录账号是否来自会话粘性命中。
|
||||
func markStickySessionHit(selection *AccountSelectionResult, hit bool) *AccountSelectionResult {
|
||||
if selection != nil && hit {
|
||||
selection.stickySessionHit = true
|
||||
}
|
||||
return selection
|
||||
}
|
||||
|
||||
func (s *OpenAIGatewayService) schedulingConfig() config.GatewaySchedulingConfig {
|
||||
if s.cfg != nil {
|
||||
return s.cfg.Gateway.Scheduling
|
||||
|
||||
@@ -0,0 +1,265 @@
|
||||
//go:build unit
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func newLegacySchedulerDecisionTestAccounts(groupID int64, secondWSEnabled bool) []Account {
|
||||
return []Account{
|
||||
{
|
||||
ID: 38101, Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
|
||||
Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0,
|
||||
GroupIDs: []int64{groupID},
|
||||
Extra: map[string]any{"openai_apikey_responses_websockets_v2_enabled": true},
|
||||
},
|
||||
{
|
||||
ID: 38102, Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
|
||||
Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5,
|
||||
GroupIDs: []int64{groupID},
|
||||
Extra: map[string]any{"openai_apikey_responses_websockets_v2_enabled": secondWSEnabled},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func newLegacySchedulerDecisionTestService(accounts []Account, loadBatch bool, concurrency schedulerTestConcurrencyCache) *OpenAIGatewayService {
|
||||
cfg := newSchedulerTestOpenAIWSV2Config()
|
||||
cfg.Gateway.Scheduling.LoadBatchEnabled = loadBatch
|
||||
return &OpenAIGatewayService{
|
||||
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
|
||||
cache: &schedulerTestGatewayCache{},
|
||||
cfg: cfg,
|
||||
concurrencyService: NewConcurrencyService(concurrency),
|
||||
}
|
||||
}
|
||||
|
||||
func releaseLegacySchedulerDecisionSelection(selection *AccountSelectionResult) {
|
||||
if selection != nil && selection.ReleaseFunc != nil {
|
||||
selection.ReleaseFunc()
|
||||
}
|
||||
}
|
||||
|
||||
func TestLegacySchedulerDecision_StickySessionLayer(t *testing.T) {
|
||||
resetOpenAIAdvancedSchedulerSettingCacheForTest()
|
||||
ctx := context.Background()
|
||||
groupID := int64(38100)
|
||||
for _, loadBatch := range []bool{true, false} {
|
||||
name := "load_batch"
|
||||
if !loadBatch {
|
||||
name = "priority_lru"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), loadBatch, schedulerTestConcurrencyCache{})
|
||||
sessionHash := "legacy-sticky-session"
|
||||
require.NoError(t, svc.setStickySessionAccountID(ctx, &groupID, sessionHash, 38102, time.Hour))
|
||||
|
||||
selection, decision, err := svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, "", sessionHash, "gpt-5.1", nil,
|
||||
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, false, true,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, selection)
|
||||
require.NotNil(t, selection.Account)
|
||||
releaseLegacySchedulerDecisionSelection(selection)
|
||||
require.Equal(t, int64(38102), selection.Account.ID, "sticky binding must win over priority order")
|
||||
require.Equal(t, openAIAccountScheduleLayerSessionSticky, decision.Layer)
|
||||
require.True(t, decision.StickySessionHit)
|
||||
require.False(t, decision.StickyPreviousHit)
|
||||
require.Equal(t, int64(38102), decision.SelectedAccountID)
|
||||
require.Equal(t, AccountTypeAPIKey, decision.SelectedAccountType)
|
||||
|
||||
selection, decision, err = svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, "", "", "gpt-5.1", nil,
|
||||
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, false, true,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, selection)
|
||||
require.NotNil(t, selection.Account)
|
||||
releaseLegacySchedulerDecisionSelection(selection)
|
||||
require.Equal(t, int64(38101), selection.Account.ID)
|
||||
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
|
||||
require.False(t, decision.StickySessionHit)
|
||||
require.Equal(t, int64(38101), decision.SelectedAccountID)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestLegacySchedulerDecision_PreviousResponseRouting(t *testing.T) {
|
||||
resetOpenAIAdvancedSchedulerSettingCacheForTest()
|
||||
ctx := context.Background()
|
||||
groupID := int64(38110)
|
||||
responseID := "resp_legacy_route"
|
||||
|
||||
t.Run("routes to the account owning the response", func(t *testing.T) {
|
||||
svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), true, schedulerTestConcurrencyCache{})
|
||||
require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour))
|
||||
|
||||
selection, decision, err := svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, responseID, "legacy-prev-session", "gpt-5.1", nil,
|
||||
OpenAIUpstreamTransportResponsesWebsocketV2Ingress, OpenAIEndpointCapabilityChatCompletions, false, true, true,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, selection)
|
||||
require.NotNil(t, selection.Account)
|
||||
releaseLegacySchedulerDecisionSelection(selection)
|
||||
require.Equal(t, int64(38102), selection.Account.ID, "previous_response_id must route to the owning account")
|
||||
require.Equal(t, openAIAccountScheduleLayerPreviousResponse, decision.Layer)
|
||||
require.True(t, decision.StickyPreviousHit)
|
||||
require.False(t, decision.StickySessionHit)
|
||||
require.Equal(t, int64(38102), decision.SelectedAccountID)
|
||||
})
|
||||
|
||||
t.Run("excluded owner falls back to load balance", func(t *testing.T) {
|
||||
svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), true, schedulerTestConcurrencyCache{})
|
||||
require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour))
|
||||
|
||||
selection, decision, err := svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, responseID, "", "gpt-5.1", map[int64]struct{}{38102: {}},
|
||||
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, selection)
|
||||
require.NotNil(t, selection.Account)
|
||||
releaseLegacySchedulerDecisionSelection(selection)
|
||||
require.Equal(t, int64(38101), selection.Account.ID)
|
||||
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
|
||||
require.False(t, decision.StickyPreviousHit)
|
||||
})
|
||||
|
||||
t.Run("transport-incompatible owner releases its slot and falls back", func(t *testing.T) {
|
||||
released := make([]int64, 0)
|
||||
svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, false), true, schedulerTestConcurrencyCache{releasedIDs: &released})
|
||||
require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour))
|
||||
|
||||
selection, decision, err := svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, responseID, "", "gpt-5.1", nil,
|
||||
OpenAIUpstreamTransportResponsesWebsocketV2Ingress, OpenAIEndpointCapabilityChatCompletions, false, true, true,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, selection)
|
||||
require.NotNil(t, selection.Account)
|
||||
releaseLegacySchedulerDecisionSelection(selection)
|
||||
require.Equal(t, int64(38101), selection.Account.ID)
|
||||
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
|
||||
require.False(t, decision.StickyPreviousHit)
|
||||
require.Contains(t, released, int64(38102), "the owning account's slot must be released after the transport check fails")
|
||||
})
|
||||
|
||||
t.Run("channel-restricted request model fails before routing", func(t *testing.T) {
|
||||
released := make([]int64, 0)
|
||||
svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), true, schedulerTestConcurrencyCache{releasedIDs: &released})
|
||||
svc.channelService = newTestChannelService(makeStandardRepo(Channel{
|
||||
ID: 38111,
|
||||
Status: StatusActive,
|
||||
GroupIDs: []int64{groupID},
|
||||
RestrictModels: true,
|
||||
BillingModelSource: BillingModelSourceChannelMapped,
|
||||
ModelPricing: []ChannelModelPricing{{Platform: PlatformOpenAI, Models: []string{"gpt-4o"}}},
|
||||
ModelMapping: map[string]map[string]string{PlatformOpenAI: {"gpt-5.1": "o3-mini"}},
|
||||
}, map[int64]string{groupID: PlatformOpenAI}))
|
||||
require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour))
|
||||
|
||||
selection, _, err := svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, responseID, "", "gpt-5.1", nil,
|
||||
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true,
|
||||
)
|
||||
require.ErrorIs(t, err, ErrNoAvailableAccounts)
|
||||
require.Contains(t, err.Error(), "channel pricing restriction")
|
||||
require.Nil(t, selection)
|
||||
require.Empty(t, released, "the owning account's slot must not be touched for a channel-restricted model")
|
||||
})
|
||||
|
||||
t.Run("channel-restricted upstream model of the owner releases its slot and falls back", func(t *testing.T) {
|
||||
released := make([]int64, 0)
|
||||
accounts := newLegacySchedulerDecisionTestAccounts(groupID, true)
|
||||
accounts[1].Credentials = map[string]any{"model_mapping": map[string]any{"gpt-5.1": "o3-mini"}}
|
||||
svc := newLegacySchedulerDecisionTestService(accounts, true, schedulerTestConcurrencyCache{releasedIDs: &released})
|
||||
svc.channelService = newTestChannelService(makeStandardRepo(Channel{
|
||||
ID: 38112,
|
||||
Status: StatusActive,
|
||||
GroupIDs: []int64{groupID},
|
||||
RestrictModels: true,
|
||||
BillingModelSource: BillingModelSourceUpstream,
|
||||
ModelPricing: []ChannelModelPricing{{Platform: PlatformOpenAI, Models: []string{"gpt-5.1"}}},
|
||||
}, map[int64]string{groupID: PlatformOpenAI}))
|
||||
require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour))
|
||||
|
||||
selection, decision, err := svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, responseID, "", "gpt-5.1", nil,
|
||||
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, selection)
|
||||
require.NotNil(t, selection.Account)
|
||||
releaseLegacySchedulerDecisionSelection(selection)
|
||||
require.Equal(t, int64(38101), selection.Account.ID)
|
||||
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
|
||||
require.False(t, decision.StickyPreviousHit)
|
||||
require.Contains(t, released, int64(38102), "the owning account's slot must be released after the upstream model restriction check fails")
|
||||
})
|
||||
|
||||
t.Run("quarantined owner proxy releases its slot and falls back", func(t *testing.T) {
|
||||
released := make([]int64, 0)
|
||||
healthyProxy, quarantinedProxy := int64(38113), int64(38114)
|
||||
accounts := newLegacySchedulerDecisionTestAccounts(groupID, true)
|
||||
accounts[0].ProxyID = &healthyProxy
|
||||
accounts[1].ProxyID = &quarantinedProxy
|
||||
svc := newLegacySchedulerDecisionTestService(accounts, false, schedulerTestConcurrencyCache{releasedIDs: &released})
|
||||
svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
|
||||
failureThreshold: 1,
|
||||
failureWindow: time.Minute,
|
||||
quarantineTTL: 10 * time.Minute,
|
||||
maxEntries: 16,
|
||||
})
|
||||
tripped, _ := svc.openaiProxyStreamCircuit.recordFailure(quarantinedProxy, time.Now())
|
||||
require.True(t, tripped)
|
||||
require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour))
|
||||
|
||||
selection, decision, err := svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, responseID, "", "gpt-5.1", nil,
|
||||
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, selection)
|
||||
require.NotNil(t, selection.Account)
|
||||
releaseLegacySchedulerDecisionSelection(selection)
|
||||
require.Equal(t, int64(38101), selection.Account.ID)
|
||||
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
|
||||
require.False(t, decision.StickyPreviousHit)
|
||||
require.Contains(t, released, int64(38102), "the owning account's slot must be released after the proxy quarantine check fails")
|
||||
})
|
||||
|
||||
t.Run("all proxies quarantined fails open to the owner", func(t *testing.T) {
|
||||
proxy := int64(38115)
|
||||
accounts := newLegacySchedulerDecisionTestAccounts(groupID, true)
|
||||
accounts[0].ProxyID = &proxy
|
||||
accounts[1].ProxyID = &proxy
|
||||
svc := newLegacySchedulerDecisionTestService(accounts, false, schedulerTestConcurrencyCache{})
|
||||
svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
|
||||
failureThreshold: 1,
|
||||
failureWindow: time.Minute,
|
||||
quarantineTTL: 10 * time.Minute,
|
||||
maxEntries: 16,
|
||||
})
|
||||
tripped, _ := svc.openaiProxyStreamCircuit.recordFailure(proxy, time.Now())
|
||||
require.True(t, tripped)
|
||||
require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour))
|
||||
|
||||
selection, decision, err := svc.SelectAccountWithSchedulerForCapability(
|
||||
ctx, &groupID, responseID, "", "gpt-5.1", nil,
|
||||
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true,
|
||||
)
|
||||
require.NoError(t, err, "quarantine must fail open instead of returning no available accounts")
|
||||
require.NotNil(t, selection)
|
||||
require.NotNil(t, selection.Account)
|
||||
releaseLegacySchedulerDecisionSelection(selection)
|
||||
require.Equal(t, int64(38102), selection.Account.ID)
|
||||
require.Equal(t, openAIAccountScheduleLayerPreviousResponse, decision.Layer)
|
||||
require.True(t, decision.StickyPreviousHit)
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user