From 0d0200059223d66c746d32e0e4980ca822cf8762 Mon Sep 17 00:00:00 2001 From: xlplbo Date: Sun, 20 Sep 2026 19:28:13 +0800 Subject: [PATCH] =?UTF-8?q?fix(openai):=20=E9=9D=9E=E9=AB=98=E7=BA=A7?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E6=8C=89=20previous=5Fresponse=20=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=E5=88=B0=E6=8C=81=E6=9C=89=E5=93=8D=E5=BA=94=E7=9A=84?= =?UTF-8?q?=E8=B4=A6=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/openai_account_scheduler.go | 63 +++++++ .../openai_legacy_scheduler_decision_test.go | 178 ++++++++++++++++++ 2 files changed, 241 insertions(+) diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 0085182333..c136f8c3fb 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -2235,6 +2235,60 @@ func applyLegacySelectionDecision(decision *OpenAIAccountScheduleDecision, selec } } +// 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, @@ -2271,6 +2325,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) diff --git a/backend/internal/service/openai_legacy_scheduler_decision_test.go b/backend/internal/service/openai_legacy_scheduler_decision_test.go index 9e30512a61..2e1ac52176 100644 --- a/backend/internal/service/openai_legacy_scheduler_decision_test.go +++ b/backend/internal/service/openai_legacy_scheduler_decision_test.go @@ -1,3 +1,5 @@ +//go:build unit + package service import ( @@ -43,6 +45,7 @@ func releaseLegacySchedulerDecisionSelection(selection *AccountSelectionResult) } func TestLegacySchedulerDecision_StickySessionLayer(t *testing.T) { + resetOpenAIAdvancedSchedulerSettingCacheForTest() ctx := context.Background() groupID := int64(38100) for _, loadBatch := range []bool{true, false} { @@ -85,3 +88,178 @@ func TestLegacySchedulerDecision_StickySessionLayer(t *testing.T) { }) } } + +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) + }) +}