fix(openai): 非高级调度按 previous_response 路由到持有响应的账号

2026-10-21 18:06:19 +00:00 · 2026-09-20 19:28:13 +08:00
parent 8d88b66668
commit 0d02000592
2 changed files with 241 additions and 0 deletions
@@ -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)
@@ -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)
+	})
+}