diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 97e35fdb1b..0b6094fcf8 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -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 重放后才能在 // 调度栈之外做抢槽后终检与准入后粘性绑定。 diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 28e30dfadf..0085182333 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -2221,6 +2221,20 @@ 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 + } +} + func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce( ctx context.Context, groupID *int64, @@ -2302,6 +2316,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 { @@ -2328,6 +2343,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 { diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index 859e782291..2f9a750caa 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -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 diff --git a/backend/internal/service/openai_legacy_scheduler_decision_test.go b/backend/internal/service/openai_legacy_scheduler_decision_test.go new file mode 100644 index 0000000000..9e30512a61 --- /dev/null +++ b/backend/internal/service/openai_legacy_scheduler_decision_test.go @@ -0,0 +1,87 @@ +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) { + 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) + }) + } +}