diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index e9ddb39b01..95bc3a0580 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -917,6 +917,8 @@ type GatewayConfig struct { OpenAIScheduler GatewayOpenAISchedulerConfig `mapstructure:"openai_scheduler"` // OpenAIHTTP2: OpenAI HTTP 上游协议策略（默认启用 HTTP/2，可按代理能力回退 HTTP/1.1） OpenAIHTTP2 GatewayOpenAIHTTP2Config `mapstructure:"openai_http2"` + // OpenAIProxyStreamCircuit: Responses SSE 代理断流熔断策略。 + OpenAIProxyStreamCircuit GatewayOpenAIProxyStreamCircuitConfig `mapstructure:"openai_proxy_stream_circuit"` // ImageConcurrency: 图片生成独立并发限制配置（默认关闭） ImageConcurrency ImageConcurrencyConfig `mapstructure:"image_concurrency"` @@ -1012,6 +1014,17 @@ type GatewayOpenAIHTTP2Config struct { FallbackTTLSeconds int `mapstructure:"fallback_ttl_seconds"` } +// GatewayOpenAIProxyStreamCircuitConfig controls the bounded, in-process +// proxy-ID circuit used for incomplete OpenAI Responses SSE streams. +type GatewayOpenAIProxyStreamCircuitConfig struct { + // FailureThreshold: 统计窗口内多少次断流后隔离代理。 + FailureThreshold int `mapstructure:"failure_threshold"` + // WindowSeconds: 断流统计窗口（秒）。 + WindowSeconds int `mapstructure:"window_seconds"` + // TTLSeconds: 代理隔离持续时间（秒）。 + TTLSeconds int `mapstructure:"ttl_seconds"` +} + // UserMessageQueueConfig 用户消息串行队列配置 // 用于 Anthropic OAuth/SetupToken 账号的用户消息串行化发送 type UserMessageQueueConfig struct { @@ -2238,6 +2251,9 @@ func setDefaults() { viper.SetDefault("gateway.openai_http2.fallback_error_threshold", 2) viper.SetDefault("gateway.openai_http2.fallback_window_seconds", 60) viper.SetDefault("gateway.openai_http2.fallback_ttl_seconds", 600) + viper.SetDefault("gateway.openai_proxy_stream_circuit.failure_threshold", 2) + viper.SetDefault("gateway.openai_proxy_stream_circuit.window_seconds", 60) + viper.SetDefault("gateway.openai_proxy_stream_circuit.ttl_seconds", 600) viper.SetDefault("gateway.image_concurrency.enabled", false) viper.SetDefault("gateway.image_concurrency.max_concurrent_requests", 0) viper.SetDefault("gateway.image_concurrency.overflow_mode", ImageConcurrencyOverflowModeReject) @@ -3231,6 +3247,15 @@ func (c *Config) Validate() error { if c.Gateway.OpenAIHTTP2.FallbackTTLSeconds < 0 { return fmt.Errorf("gateway.openai_http2.fallback_ttl_seconds must be non-negative") } + if c.Gateway.OpenAIProxyStreamCircuit.FailureThreshold < 0 { + return fmt.Errorf("gateway.openai_proxy_stream_circuit.failure_threshold must be non-negative") + } + if c.Gateway.OpenAIProxyStreamCircuit.WindowSeconds < 0 { + return fmt.Errorf("gateway.openai_proxy_stream_circuit.window_seconds must be non-negative") + } + if c.Gateway.OpenAIProxyStreamCircuit.TTLSeconds < 0 { + return fmt.Errorf("gateway.openai_proxy_stream_circuit.ttl_seconds must be non-negative") + } weights := c.Gateway.OpenAIWS.SchedulerScoreWeights for _, weight := range []float64{ weights.Priority, weights.Load, weights.Queue, weights.ErrorRate, weights.TTFT, diff --git a/backend/internal/config/config_test.go b/backend/internal/config/config_test.go index 0c55d2776d..e4186732a8 100644 --- a/backend/internal/config/config_test.go +++ b/backend/internal/config/config_test.go @@ -497,6 +497,22 @@ func TestLoadDefaultOpenAIHTTP2Enabled(t *testing.T) { require.NoError(t, err) require.True(t, cfg.Gateway.OpenAIHTTP2.Enabled) require.True(t, cfg.Gateway.OpenAIHTTP2.AllowProxyFallbackToHTTP1) + require.Equal(t, 2, cfg.Gateway.OpenAIProxyStreamCircuit.FailureThreshold) + require.Equal(t, 60, cfg.Gateway.OpenAIProxyStreamCircuit.WindowSeconds) + require.Equal(t, 600, cfg.Gateway.OpenAIProxyStreamCircuit.TTLSeconds) +} + +func TestLoadOpenAIProxyStreamCircuitFromEnv(t *testing.T) { + resetViperWithJWTSecret(t) + t.Setenv("GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD", "3") + t.Setenv("GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS", "90") + t.Setenv("GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS", "420") + + cfg, err := Load() + require.NoError(t, err) + require.Equal(t, 3, cfg.Gateway.OpenAIProxyStreamCircuit.FailureThreshold) + require.Equal(t, 90, cfg.Gateway.OpenAIProxyStreamCircuit.WindowSeconds) + require.Equal(t, 420, cfg.Gateway.OpenAIProxyStreamCircuit.TTLSeconds) } func TestLoadOpenAIHTTP2DisabledFromEnv(t *testing.T) { @@ -1744,6 +1760,21 @@ func TestValidateConfigErrors(t *testing.T) { mutate: func(c *Config) { c.Gateway.OpenAIHTTP2.FallbackTTLSeconds = -1 }, wantErr: "gateway.openai_http2.fallback_ttl_seconds", }, + { + name: "gateway openai proxy stream circuit threshold", + mutate: func(c *Config) { c.Gateway.OpenAIProxyStreamCircuit.FailureThreshold = -1 }, + wantErr: "gateway.openai_proxy_stream_circuit.failure_threshold", + }, + { + name: "gateway openai proxy stream circuit window", + mutate: func(c *Config) { c.Gateway.OpenAIProxyStreamCircuit.WindowSeconds = -1 }, + wantErr: "gateway.openai_proxy_stream_circuit.window_seconds", + }, + { + name: "gateway openai proxy stream circuit ttl", + mutate: func(c *Config) { c.Gateway.OpenAIProxyStreamCircuit.TTLSeconds = -1 }, + wantErr: "gateway.openai_proxy_stream_circuit.ttl_seconds", + }, { name: "gateway stream data interval range", mutate: func(c *Config) { c.Gateway.StreamDataIntervalTimeout = 5 }, diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 3be5aabc57..870a6e0ef3 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -1678,6 +1678,9 @@ func (s *defaultOpenAIAccountScheduler) isAccountRequestCompatibleReason(ctx con if s != nil && s.service != nil && s.service.isOpenAIAccountRequestRuntimeBlocked(account, req.RequestedModel) { return false, "runtime_blocked" } + if s != nil && s.service != nil && s.service.isOpenAIProxyStreamQuarantined(account) { + return false, "proxy_stream_quarantined" + } // Quota auto-pause must be evaluated during the initial filter too. Without it the // TopK candidate pool can be filled with paused accounts and the later fresh/DB // rechecks won't reach healthy accounts that fell outside TopK — manifesting as diff --git a/backend/internal/service/openai_account_scheduler_test.go b/backend/internal/service/openai_account_scheduler_test.go index f9735934b5..9ffa0799bb 100644 --- a/backend/internal/service/openai_account_scheduler_test.go +++ b/backend/internal/service/openai_account_scheduler_test.go @@ -1441,6 +1441,40 @@ func TestOpenAIGatewayService_OpenAIAccountSchedulerMetrics_DisabledNoOp(t *test require.Equal(t, OpenAIAccountSchedulerMetricsSnapshot{}, snapshot) } +func TestOpenAIGatewayService_SelectAccountWithScheduler_SkipsQuarantinedSharedProxy(t *testing.T) { + resetOpenAIAdvancedSchedulerSettingCacheForTest() + + proxyA := int64(4698) + proxyB := int64(4699) + accounts := []Account{ + {ID: 469801, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0, ProxyID: &proxyA}, + {ID: 469802, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0, ProxyID: &proxyA}, + {ID: 469803, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5, ProxyID: &proxyB}, + } + cfg := &config.Config{} + cfg.Gateway.Scheduling.LoadBatchEnabled = false + svc := &OpenAIGatewayService{ + accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts}, + cfg: cfg, + concurrencyService: NewConcurrencyService(schedulerTestConcurrencyCache{}), + openaiProxyStreamCircuit: newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 1, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 16, + }), + } + svc.openaiProxyStreamCircuit.recordFailure(proxyA, time.Now()) + + selection, _, err := svc.SelectAccountWithScheduler( + context.Background(), nil, "", "", "gpt-5.6-sol", nil, OpenAIUpstreamTransportAny, false, + ) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + require.Equal(t, int64(469803), selection.Account.ID) +} + func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyRateLimitedAccountFallsBackToFreshCandidate(t *testing.T) { ctx := context.Background() groupID := int64(10101) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 1fb281b4a2..d440c27766 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -1166,7 +1166,8 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( } } if err := documentScanner.Err(); err != nil { - if sawTerminalEvent && !sawFailedEvent { + if (sawDone || sawTerminalEvent) && !sawFailedEvent { + s.clearOpenAIProxyStreamDisconnect(account) return resultWithUsage(), nil } if sawFailedEvent { @@ -1190,6 +1191,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( if clientDisconnected { return resultWithUsage(), fmt.Errorf("stream usage incomplete after disconnect: %w", err) } + s.recordOpenAIProxyStreamDisconnect(account, err, upstreamRequestID) logger.LegacyPrintf("service.openai_gateway", "[OpenAI passthrough] 流读取异常中断: account=%d request_id=%s err=%v", account.ID, @@ -1211,8 +1213,12 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( return resultWithUsage(), s.newOpenAIStreamFailoverError(c, account, true, upstreamRequestID, nil, "OpenAI stream ended before a terminal event") } + s.recordOpenAIProxyStreamDisconnect(account, errors.New("stream ended before terminal event"), upstreamRequestID) return resultWithUsage(), errors.New("stream usage incomplete: missing terminal event") } + if (sawDone || sawTerminalEvent) && !sawFailedEvent { + s.clearOpenAIProxyStreamDisconnect(account) + } return resultWithUsage(), nil } diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 12923274c6..7c10724028 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -314,6 +314,9 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. // EOF dispatches the final SSE event even without a trailing blank line. completeGuardedEvent(true) } + if sawTerminalEvent && !sawFailedEvent { + s.clearOpenAIProxyStreamDisconnect(account) + } if !sawTerminalEvent && !openAIStreamClientOutputStarted(c, clientOutputStarted) && !eventShouldFlush { return resultWithUsage(), s.newOpenAIStreamFailoverError( c, @@ -326,6 +329,9 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } flushPending("Client disconnected during final flush, returning collected usage") if !sawTerminalEvent { + if openAIStreamClientOutputStarted(c, clientOutputStarted) && !clientDisconnected { + s.recordOpenAIProxyStreamDisconnect(account, errors.New("stream ended before terminal event"), upstreamRequestID) + } return resultWithUsage(), fmt.Errorf("stream usage incomplete: missing terminal event") } if sawFailedEvent { @@ -357,6 +363,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } if sawTerminalEvent { if !sawFailedEvent { + s.clearOpenAIProxyStreamDisconnect(account) logger.LegacyPrintf("service.openai_gateway", "Upstream scan ended after terminal event: %v", scanErr) } result, err := finalizeStream() @@ -386,6 +393,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. if clientDisconnected { return resultWithUsage(), fmt.Errorf("stream usage incomplete after disconnect: %w", scanErr), true } + s.recordOpenAIProxyStreamDisconnect(account, scanErr, upstreamRequestID) sendErrorEvent("stream_read_error") return resultWithUsage(), fmt.Errorf("stream read error: %w", scanErr), true } diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index ba350f5062..a895b2ebf5 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -1241,6 +1241,9 @@ func (s *OpenAIGatewayService) resolveFreshSchedulableOpenAIAccount(ctx context. if s.isOpenAIAccountRequestRuntimeBlocked(fresh, requestedModel) { return nil } + if s.isOpenAIProxyStreamQuarantined(fresh) { + return nil + } return fresh } @@ -1270,6 +1273,9 @@ func (s *OpenAIGatewayService) recheckSelectedOpenAIAccountFromDB(ctx context.Co if !parentHealthyForShadow(account, s.parentAccountLookup(ctx)) { return nil } + if s.isOpenAIProxyStreamQuarantined(account) { + return nil + } return account } @@ -1289,6 +1295,9 @@ func (s *OpenAIGatewayService) recheckSelectedOpenAIAccountFromDB(ctx context.Co if s.isOpenAIAccountRequestRuntimeBlocked(latest, requestedModel) { return nil } + if s.isOpenAIProxyStreamQuarantined(latest) { + return nil + } return latest } diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index b0040800fa..5cb43f7baa 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -414,6 +414,7 @@ type OpenAIGatewayService struct { openaiWSPoolOnce sync.Once openaiWSStateStoreOnce sync.Once openaiSchedulerOnce sync.Once + openaiProxyStreamCircuitOnce sync.Once openaiWSPassthroughDialerOnce sync.Once openaiModelTransientOnce sync.Once agentIdentityTaskMu sync.Mutex @@ -423,6 +424,7 @@ type OpenAIGatewayService struct { openaiWSPassthroughDialer openAIWSClientDialer openaiAccountStats *openAIAccountRuntimeStats openaiModelTransient *openAIAccountModelTransientState + openaiProxyStreamCircuit *openAIProxyStreamCircuit openaiWSFallbackUntil sync.Map // key: int64(accountID), value: time.Time openaiAccountRuntimeBlockUntil sync.Map // key: int64(accountID), value: time.Time diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index 2c2725ea78..d8cfba73b8 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -125,6 +125,20 @@ type errReadCloser struct { func (r errReadCloser) Read([]byte) (int, error) { return 0, r.err } func (r errReadCloser) Close() error { return nil } +type openAIStreamReadThenErrorCloser struct { + reader *strings.Reader + err error +} + +func (r *openAIStreamReadThenErrorCloser) Read(p []byte) (int, error) { + if r.reader != nil && r.reader.Len() > 0 { + return r.reader.Read(p) + } + return 0, r.err +} + +func (r *openAIStreamReadThenErrorCloser) Close() error { return nil } + type failingGinWriter struct { gin.ResponseWriter failAfter int @@ -1329,6 +1343,101 @@ func TestOpenAIStreamingReadErrorBeforeOutputReturnsFailover(t *testing.T) { require.Empty(t, rec.Body.String()) } +func TestOpenAIStreamingPostOutputDisconnectQuarantinesSharedProxyWithoutSameStreamFailover(t *testing.T) { + gin.SetMode(gin.TestMode) + proxyID := int64(4698) + account := &Account{ + ID: 469801, + Name: "oauth-on-shared-proxy", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + ProxyID: &proxyID, + } + svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{ + MaxLineSize: defaultMaxLineSize, + }}} + + for _, readErr := range []error{ + io.ErrUnexpectedEOF, + errors.New("http2: client connection lost"), + } { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: &openAIStreamReadThenErrorCloser{ + reader: strings.NewReader(strings.Join([]string{ + "event: response.output_text.delta", + `data: {"type":"response.output_text.delta","delta":"partial"}`, + "", + }, "\n")), + err: readErr, + }, + Header: http.Header{"X-Request-Id": []string{"rid-proxy-disconnect"}}, + } + + _, err := svc.handleStreamingResponse(c.Request.Context(), resp, c, account, time.Now(), "gpt-5.6-sol", "gpt-5.6-sol") + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr), "post-output disconnect must not fail over inside the same stream") + require.Contains(t, rec.Body.String(), "partial") + } + + scheduler := &defaultOpenAIAccountScheduler{service: svc} + compatible, reason := scheduler.isAccountRequestCompatibleReason(context.Background(), account, OpenAIAccountScheduleRequest{}) + require.False(t, compatible, "the next request must exclude accounts sharing the quarantined proxy") + require.Equal(t, "proxy_stream_quarantined", reason) +} + +func TestOpenAIStreamingTerminalAndClientCancellationDoNotQuarantineProxy(t *testing.T) { + gin.SetMode(gin.TestMode) + proxyID := int64(4699) + account := &Account{ID: 469901, Name: "oauth", Platform: PlatformOpenAI, Type: AccountTypeOAuth, ProxyID: &proxyID} + svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}}} + + terminalRecorder := httptest.NewRecorder() + terminalCtx, _ := gin.CreateTestContext(terminalRecorder) + terminalCtx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + terminalResp := &http.Response{ + StatusCode: http.StatusOK, + Body: &openAIStreamReadThenErrorCloser{ + reader: strings.NewReader(strings.Join([]string{ + "event: response.completed", + `data: {"type":"response.completed","response":{"status":"completed","output":[]}}`, + "", + }, "\n")), + err: io.ErrUnexpectedEOF, + }, + Header: http.Header{}, + } + _, err := svc.handleStreamingResponse(terminalCtx.Request.Context(), terminalResp, terminalCtx, account, time.Now(), "model", "model") + require.NoError(t, err) + + for range 2 { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil).WithContext(ctx) + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: &openAIStreamReadThenErrorCloser{ + reader: strings.NewReader("data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"), + err: context.Canceled, + }, + Header: http.Header{}, + } + _, err = svc.handleStreamingResponse(c.Request.Context(), resp, c, account, time.Now(), "model", "model") + require.Error(t, err) + } + + scheduler := &defaultOpenAIAccountScheduler{service: svc} + compatible, reason := scheduler.isAccountRequestCompatibleReason(context.Background(), account, OpenAIAccountScheduleRequest{}) + require.True(t, compatible) + require.Empty(t, reason) +} + func TestOpenAIStreamingResponseFailedBeforeOutputReturnsFailover(t *testing.T) { gin.SetMode(gin.TestMode) cfg := &config.Config{ @@ -1894,6 +2003,35 @@ func TestOpenAIStreamingPassthroughMissingTerminalEventReturnsIncompleteError(t } } +func TestOpenAIStreamingPassthroughPostOutputDisconnectQuarantinesSharedProxy(t *testing.T) { + gin.SetMode(gin.TestMode) + proxyID := int64(4698) + account := &Account{ID: 469804, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, ProxyID: &proxyID} + svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}}} + + for _, readErr := range []error{io.ErrUnexpectedEOF, errors.New("http2: client connection lost")} { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: &openAIStreamReadThenErrorCloser{ + reader: strings.NewReader("data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"), + err: readErr, + }, + Header: http.Header{"X-Request-Id": []string{"rid-passthrough-proxy-disconnect"}}, + } + + _, err := svc.handleStreamingResponsePassthrough(c.Request.Context(), resp, c, account, time.Now(), "model", "model") + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr), "post-output disconnect must not fail over inside the same stream") + require.Contains(t, rec.Body.String(), "partial") + } + + require.True(t, svc.isOpenAIProxyStreamQuarantined(account)) +} + func TestOpenAIStreamingPassthroughResponseFailedBeforeOutputReturnsFailover(t *testing.T) { gin.SetMode(gin.TestMode) cfg := &config.Config{ diff --git a/backend/internal/service/openai_proxy_stream_circuit.go b/backend/internal/service/openai_proxy_stream_circuit.go new file mode 100644 index 0000000000..a28a005143 --- /dev/null +++ b/backend/internal/service/openai_proxy_stream_circuit.go @@ -0,0 +1,229 @@ +package service + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/logger" + "go.uber.org/zap" +) + +const ( + defaultOpenAIProxyStreamFailureThreshold = 2 + defaultOpenAIProxyStreamFailureWindow = time.Minute + defaultOpenAIProxyStreamQuarantineTTL = 10 * time.Minute + defaultOpenAIProxyStreamCircuitMaxEntries = 4096 +) + +type openAIProxyStreamCircuitSettings struct { + failureThreshold int + failureWindow time.Duration + quarantineTTL time.Duration + maxEntries int +} + +type openAIProxyStreamCircuitEntry struct { + failureCount int + windowStart time.Time + blockedUntil time.Time + lastTouched time.Time +} + +// openAIProxyStreamCircuit is an in-process, proxy-ID keyed circuit. It is +// intentionally bounded and ephemeral: a restart clears observations, while a +// tripped entry expires automatically after its TTL. +type openAIProxyStreamCircuit struct { + mu sync.Mutex + settings openAIProxyStreamCircuitSettings + entries map[int64]openAIProxyStreamCircuitEntry +} + +func resolveOpenAIProxyStreamCircuitSettings(s *OpenAIGatewayService) openAIProxyStreamCircuitSettings { + settings := openAIProxyStreamCircuitSettings{ + failureThreshold: defaultOpenAIProxyStreamFailureThreshold, + failureWindow: defaultOpenAIProxyStreamFailureWindow, + quarantineTTL: defaultOpenAIProxyStreamQuarantineTTL, + maxEntries: defaultOpenAIProxyStreamCircuitMaxEntries, + } + if s == nil || s.cfg == nil { + return settings + } + cfg := s.cfg.Gateway.OpenAIProxyStreamCircuit + if cfg.FailureThreshold > 0 { + settings.failureThreshold = cfg.FailureThreshold + } + if cfg.WindowSeconds > 0 { + settings.failureWindow = time.Duration(cfg.WindowSeconds) * time.Second + } + if cfg.TTLSeconds > 0 { + settings.quarantineTTL = time.Duration(cfg.TTLSeconds) * time.Second + } + return settings +} + +func newOpenAIProxyStreamCircuit(settings openAIProxyStreamCircuitSettings) *openAIProxyStreamCircuit { + if settings.failureThreshold <= 0 { + settings.failureThreshold = defaultOpenAIProxyStreamFailureThreshold + } + if settings.failureWindow <= 0 { + settings.failureWindow = defaultOpenAIProxyStreamFailureWindow + } + if settings.quarantineTTL <= 0 { + settings.quarantineTTL = defaultOpenAIProxyStreamQuarantineTTL + } + if settings.maxEntries <= 0 { + settings.maxEntries = defaultOpenAIProxyStreamCircuitMaxEntries + } + return &openAIProxyStreamCircuit{ + settings: settings, + entries: make(map[int64]openAIProxyStreamCircuitEntry), + } +} + +func (s *OpenAIGatewayService) getOpenAIProxyStreamCircuit() *openAIProxyStreamCircuit { + if s == nil { + return nil + } + s.openaiProxyStreamCircuitOnce.Do(func() { + if s.openaiProxyStreamCircuit == nil { + s.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(resolveOpenAIProxyStreamCircuitSettings(s)) + } + }) + return s.openaiProxyStreamCircuit +} + +func (c *openAIProxyStreamCircuit) recordFailure(proxyID int64, now time.Time) (bool, time.Time) { + if c == nil || proxyID <= 0 { + return false, time.Time{} + } + c.mu.Lock() + defer c.mu.Unlock() + + entry, exists := c.entries[proxyID] + if exists && now.Before(entry.blockedUntil) { + entry.lastTouched = now + c.entries[proxyID] = entry + return false, entry.blockedUntil + } + if !exists { + c.ensureCapacityLocked(now) + } + if entry.windowStart.IsZero() || now.Before(entry.windowStart) || now.Sub(entry.windowStart) > c.settings.failureWindow { + entry.failureCount = 0 + entry.windowStart = now + entry.blockedUntil = time.Time{} + } + entry.failureCount++ + entry.lastTouched = now + tripped := entry.failureCount >= c.settings.failureThreshold + if tripped { + entry.blockedUntil = now.Add(c.settings.quarantineTTL) + } + c.entries[proxyID] = entry + return tripped, entry.blockedUntil +} + +func (c *openAIProxyStreamCircuit) recordSuccess(proxyID int64) bool { + if c == nil || proxyID <= 0 { + return false + } + c.mu.Lock() + defer c.mu.Unlock() + if _, ok := c.entries[proxyID]; !ok { + return false + } + delete(c.entries, proxyID) + return true +} + +func (c *openAIProxyStreamCircuit) isBlocked(proxyID int64, now time.Time) bool { + if c == nil || proxyID <= 0 { + return false + } + c.mu.Lock() + defer c.mu.Unlock() + entry, ok := c.entries[proxyID] + if !ok || entry.blockedUntil.IsZero() { + return false + } + if !now.Before(entry.blockedUntil) { + delete(c.entries, proxyID) + return false + } + return true +} + +func (c *openAIProxyStreamCircuit) ensureCapacityLocked(now time.Time) { + if len(c.entries) < c.settings.maxEntries { + return + } + for proxyID, entry := range c.entries { + staleObservation := entry.blockedUntil.IsZero() && now.Sub(entry.lastTouched) > c.settings.failureWindow + expiredQuarantine := !entry.blockedUntil.IsZero() && !now.Before(entry.blockedUntil) + if staleObservation || expiredQuarantine { + delete(c.entries, proxyID) + } + } + if len(c.entries) < c.settings.maxEntries { + return + } + var oldestProxyID int64 + var oldest time.Time + for proxyID, entry := range c.entries { + if oldestProxyID == 0 || entry.lastTouched.Before(oldest) { + oldestProxyID = proxyID + oldest = entry.lastTouched + } + } + if oldestProxyID > 0 { + delete(c.entries, oldestProxyID) + } +} + +func openAIProxyStreamCircuitProxyID(account *Account) (int64, bool) { + if account == nil || account.Platform != PlatformOpenAI || account.ProxyID == nil || *account.ProxyID <= 0 { + return 0, false + } + return *account.ProxyID, true +} + +func (s *OpenAIGatewayService) recordOpenAIProxyStreamDisconnect(account *Account, streamErr error, upstreamRequestID string) { + proxyID, ok := openAIProxyStreamCircuitProxyID(account) + if !ok || streamErr == nil || errors.Is(streamErr, context.Canceled) || errors.Is(streamErr, context.DeadlineExceeded) { + return + } + circuit := s.getOpenAIProxyStreamCircuit() + tripped, until := circuit.recordFailure(proxyID, time.Now()) + if !tripped { + return + } + logger.L().With(zap.String("component", "service.openai_gateway")).Warn( + "openai.proxy_quarantined_stream_disconnect", + zap.Int64("proxy_id", proxyID), + zap.Int64("account_id", account.ID), + zap.Time("until", until), + zap.String("upstream_request_id", upstreamRequestID), + zap.String("error", sanitizeUpstreamErrorMessage(streamErr.Error())), + ) +} + +func (s *OpenAIGatewayService) clearOpenAIProxyStreamDisconnect(account *Account) { + proxyID, ok := openAIProxyStreamCircuitProxyID(account) + if !ok { + return + } + if circuit := s.getOpenAIProxyStreamCircuit(); circuit != nil { + circuit.recordSuccess(proxyID) + } +} + +func (s *OpenAIGatewayService) isOpenAIProxyStreamQuarantined(account *Account) bool { + proxyID, ok := openAIProxyStreamCircuitProxyID(account) + if !ok { + return false + } + circuit := s.getOpenAIProxyStreamCircuit() + return circuit != nil && circuit.isBlocked(proxyID, time.Now()) +} diff --git a/backend/internal/service/openai_proxy_stream_circuit_test.go b/backend/internal/service/openai_proxy_stream_circuit_test.go new file mode 100644 index 0000000000..c278ea8c1e --- /dev/null +++ b/backend/internal/service/openai_proxy_stream_circuit_test.go @@ -0,0 +1,56 @@ +package service + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestOpenAIProxyStreamCircuitThresholdTTLAndSuccessReset(t *testing.T) { + base := time.Unix(1_800_000_000, 0) + circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 2, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 16, + }) + + tripped, _ := circuit.recordFailure(1, base) + require.False(t, tripped) + require.False(t, circuit.isBlocked(1, base)) + require.True(t, circuit.recordSuccess(1)) + + tripped, _ = circuit.recordFailure(1, base.Add(10*time.Second)) + require.False(t, tripped, "success must clear the previous failure observation") + tripped, until := circuit.recordFailure(1, base.Add(20*time.Second)) + require.True(t, tripped) + require.Equal(t, base.Add(20*time.Second+10*time.Minute), until) + require.True(t, circuit.isBlocked(1, until.Add(-time.Nanosecond))) + require.False(t, circuit.isBlocked(1, until), "TTL expiry must re-admit the proxy") + + tripped, _ = circuit.recordFailure(2, base) + require.False(t, tripped) + tripped, _ = circuit.recordFailure(2, base.Add(2*time.Minute)) + require.False(t, tripped, "failures outside the window must not accumulate") +} + +func TestOpenAIProxyStreamCircuitBoundsEntries(t *testing.T) { + base := time.Unix(1_800_000_000, 0) + circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 1, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 2, + }) + + circuit.recordFailure(1, base) + circuit.recordFailure(2, base.Add(time.Second)) + circuit.recordFailure(3, base.Add(2*time.Second)) + + circuit.mu.Lock() + defer circuit.mu.Unlock() + require.Len(t, circuit.entries, 2) + _, oldestRetained := circuit.entries[1] + require.False(t, oldestRetained, "the oldest entry must be evicted at the bound") +} diff --git a/deploy/.env.example b/deploy/.env.example index 4ddac781e8..baf7cec2cb 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -287,6 +287,9 @@ GATEWAY_OPENAI_HTTP2_ALLOW_PROXY_FALLBACK_TO_HTTP1=true GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD=2 GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS=60 GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS=600 +GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD=2 +GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS=60 +GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS=600 # 上游连接池：每主机最大连接数（默认 1024；流式/HTTP1.1 场景可调大，如 2400/4096） GATEWAY_MAX_CONNS_PER_HOST=2048 # 上游连接池：最大空闲连接总数（默认 2560；账号/代理隔离 + 高并发场景可调大） diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index 6bcdefdceb..f66224457a 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -401,6 +401,11 @@ gateway: fallback_error_threshold: 2 fallback_window_seconds: 60 fallback_ttl_seconds: 600 + # OpenAI Responses SSE 代理断流熔断；按 proxy_id 跨账号隔离，仅影响下一次调度。 + openai_proxy_stream_circuit: + failure_threshold: 2 + window_seconds: 60 + ttl_seconds: 600 # HTTP upstream connection pool settings (HTTP/2 + multi-proxy scenario defaults) # HTTP 上游连接池配置（HTTP/2 + 多代理场景默认值） # Max idle connections across all hosts diff --git a/deploy/docker-compose.dev.yml b/deploy/docker-compose.dev.yml index 081e1a1830..db00ac9de8 100644 --- a/deploy/docker-compose.dev.yml +++ b/deploy/docker-compose.dev.yml @@ -58,6 +58,9 @@ services: - GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD=${GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD:-2} - GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS:-60} - GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS:-600} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD:-2} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS:-60} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS:-600} - GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT=${GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT:-900} - GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL=${GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL:-10} - GATEWAY_IMAGE_CONCURRENCY_ENABLED=${GATEWAY_IMAGE_CONCURRENCY_ENABLED:-false} diff --git a/deploy/docker-compose.local.yml b/deploy/docker-compose.local.yml index ea61889913..364fcc2ecc 100644 --- a/deploy/docker-compose.local.yml +++ b/deploy/docker-compose.local.yml @@ -166,6 +166,9 @@ services: - GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD=${GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD:-2} - GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS:-60} - GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS:-600} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD:-2} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS:-60} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS:-600} - GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT=${GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT:-900} - GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL=${GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL:-10} - GATEWAY_IMAGE_CONCURRENCY_ENABLED=${GATEWAY_IMAGE_CONCURRENCY_ENABLED:-false} diff --git a/deploy/docker-compose.standalone.yml b/deploy/docker-compose.standalone.yml index f7e664ed43..04a46a8dd5 100644 --- a/deploy/docker-compose.standalone.yml +++ b/deploy/docker-compose.standalone.yml @@ -113,6 +113,9 @@ services: - GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD=${GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD:-2} - GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS:-60} - GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS:-600} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD:-2} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS:-60} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS:-600} - GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT=${GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT:-900} - GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL=${GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL:-10} - GATEWAY_IMAGE_CONCURRENCY_ENABLED=${GATEWAY_IMAGE_CONCURRENCY_ENABLED:-false} diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 16cc310986..b82910fb45 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -162,6 +162,9 @@ services: - GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD=${GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD:-2} - GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS:-60} - GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS:-600} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD:-2} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS:-60} + - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS:-600} - GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT=${GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT:-900} - GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL=${GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL:-10} - GATEWAY_IMAGE_CONCURRENCY_ENABLED=${GATEWAY_IMAGE_CONCURRENCY_ENABLED:-false}