mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-07 13:08:03 +08:00
fix(openai): quarantine proxies after stream disconnects
This commit is contained in:
Heatherm Huang
@@ -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,
|
||||
|
||||
@@ -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 },
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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；账号/代理隔离 + 高并发场景可调大）
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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}
|
||||
|
||||
Reference in New Issue
Block a user