From da49ce3f290272db1c21f1d2a5def0220e86b743 Mon Sep 17 00:00:00 2001 From: shaw Date: Fri, 31 Jul 2026 10:30:30 +0800 Subject: [PATCH] fix(openai): fail open proxy stream circuit and collapse burst disconnects The proxy stream circuit introduced in v0.1.164 (#4749) removes every account behind a quarantined proxy from scheduling. When all schedulable accounts share one proxy (a common deployment), two mid-stream disconnects within a minute zeroed out capacity for 10 minutes and every request failed with 502. One HTTP/2 connection loss also killed all multiplexed streams at once, tripping the threshold from a single event. - Quarantine now degrades to a preference: when the only reason no account is available is proxy quarantine, selection retries once with the quarantine bypassed, so capacity can never reach zero. - Disconnects within 3s per proxy collapse into one failure event. - Add gateway.openai_proxy_stream_circuit.disabled escape hatch. - A completed stream still clears the quarantine immediately; TTL, thresholds and recording guards are unchanged. --- backend/internal/config/config.go | 3 + backend/internal/config/config_test.go | 3 + .../service/openai_account_scheduler.go | 44 ++++++++- .../service/openai_account_scheduler_test.go | 38 ++++++++ .../service/openai_gateway_scheduling.go | 6 +- .../service/openai_gateway_service.go | 29 +++--- .../service/openai_gateway_service_test.go | 18 +++- .../service/openai_proxy_stream_circuit.go | 97 +++++++++++++++++-- .../openai_proxy_stream_circuit_test.go | 77 +++++++++++++++ deploy/config.example.yaml | 3 + 10 files changed, 292 insertions(+), 26 deletions(-) diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 46d5c60429..e3e0ab0c2e 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -1035,6 +1035,8 @@ type GatewayOpenAIHTTP2Config struct { // GatewayOpenAIProxyStreamCircuitConfig controls the bounded, in-process // proxy-ID circuit used for incomplete OpenAI Responses SSE streams. type GatewayOpenAIProxyStreamCircuitConfig struct { + // Disabled: 完全关闭代理断流熔断（默认开启）。 + Disabled bool `mapstructure:"disabled"` // FailureThreshold: 统计窗口内多少次断流后隔离代理。 FailureThreshold int `mapstructure:"failure_threshold"` // WindowSeconds: 断流统计窗口（秒）。 @@ -2282,6 +2284,7 @@ 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.disabled", false) 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) diff --git a/backend/internal/config/config_test.go b/backend/internal/config/config_test.go index c22c0e8852..b6ade9c4b2 100644 --- a/backend/internal/config/config_test.go +++ b/backend/internal/config/config_test.go @@ -545,6 +545,7 @@ func TestLoadDefaultOpenAIHTTP2Enabled(t *testing.T) { require.NoError(t, err) require.True(t, cfg.Gateway.OpenAIHTTP2.Enabled) require.True(t, cfg.Gateway.OpenAIHTTP2.AllowProxyFallbackToHTTP1) + require.False(t, cfg.Gateway.OpenAIProxyStreamCircuit.Disabled) require.Equal(t, 2, cfg.Gateway.OpenAIProxyStreamCircuit.FailureThreshold) require.Equal(t, 60, cfg.Gateway.OpenAIProxyStreamCircuit.WindowSeconds) require.Equal(t, 600, cfg.Gateway.OpenAIProxyStreamCircuit.TTLSeconds) @@ -552,12 +553,14 @@ func TestLoadDefaultOpenAIHTTP2Enabled(t *testing.T) { func TestLoadOpenAIProxyStreamCircuitFromEnv(t *testing.T) { resetViperWithJWTSecret(t) + t.Setenv("GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_DISABLED", "true") 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.True(t, cfg.Gateway.OpenAIProxyStreamCircuit.Disabled) require.Equal(t, 3, cfg.Gateway.OpenAIProxyStreamCircuit.FailureThreshold) require.Equal(t, 90, cfg.Gateway.OpenAIProxyStreamCircuit.WindowSeconds) require.Equal(t, 420, cfg.Gateway.OpenAIProxyStreamCircuit.TTLSeconds) diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 870a6e0ef3..ceafdda40a 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -3,6 +3,7 @@ package service import ( "container/heap" "context" + "errors" "fmt" "hash/fnv" "log/slog" @@ -1678,7 +1679,7 @@ 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) { + if s != nil && s.service != nil && s.service.isOpenAIProxyStreamQuarantined(ctx, account) { return false, "proxy_stream_quarantined" } // Quota auto-pause must be evaluated during the initial filter too. Without it the @@ -2035,6 +2036,13 @@ func (s *OpenAIGatewayService) SelectAccountWithSchedulerForImages( return selection, decision, err } +// selectAccountWithScheduler wraps selectAccountWithSchedulerOnce with a +// fail-open second pass for the proxy stream circuit (#5056): when the only +// reason no account is available is that every candidate sits behind a +// quarantined proxy, the quarantine must degrade to a preference instead of +// zeroing out capacity. The retry re-runs the exact same selection with the +// quarantine checks bypassed, so healthy proxies always win the first pass +// and quarantined ones only serve when nothing else can. func (s *OpenAIGatewayService) selectAccountWithScheduler( ctx context.Context, groupID *int64, @@ -2049,6 +2057,40 @@ func (s *OpenAIGatewayService) selectAccountWithScheduler( platform string, previousResponseCanMove bool, useUpstreamTokenCost bool, +) (*AccountSelectionResult, OpenAIAccountScheduleDecision, error) { + selection, decision, err := s.selectAccountWithSchedulerOnce(ctx, groupID, previousResponseID, sessionHash, requestedModel, excludedIDs, requiredTransport, requiredCapability, requiredImageCapability, requireCompact, platform, previousResponseCanMove, useUpstreamTokenCost) + if err == nil || openAIProxyStreamQuarantineBypassed(ctx) { + return selection, decision, err + } + if !errors.Is(err, ErrNoAvailableAccounts) && !errors.Is(err, ErrNoAvailableCompactAccounts) { + return selection, decision, err + } + // The circuit only ever quarantines PlatformOpenAI accounts. + if normalizeOpenAICompatiblePlatform(platform) != PlatformOpenAI { + return selection, decision, err + } + blocked := s.getOpenAIProxyStreamCircuit().activeBlockCount(time.Now()) + if blocked == 0 { + return selection, decision, err + } + s.logOpenAIProxyStreamQuarantineFailOpen(requestedModel, blocked) + return s.selectAccountWithSchedulerOnce(withOpenAIProxyStreamQuarantineBypass(ctx), groupID, previousResponseID, sessionHash, requestedModel, excludedIDs, requiredTransport, requiredCapability, requiredImageCapability, requireCompact, platform, previousResponseCanMove, useUpstreamTokenCost) +} + +func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce( + 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, + previousResponseCanMove bool, + useUpstreamTokenCost bool, ) (*AccountSelectionResult, OpenAIAccountScheduleDecision, error) { ctx = s.withOpenAIQuotaAutoPauseContext(ctx) platform = normalizeOpenAICompatiblePlatform(platform) diff --git a/backend/internal/service/openai_account_scheduler_test.go b/backend/internal/service/openai_account_scheduler_test.go index 9ffa0799bb..a7c51f6fa7 100644 --- a/backend/internal/service/openai_account_scheduler_test.go +++ b/backend/internal/service/openai_account_scheduler_test.go @@ -1475,6 +1475,44 @@ func TestOpenAIGatewayService_SelectAccountWithScheduler_SkipsQuarantinedSharedP require.Equal(t, int64(469803), selection.Account.ID) } +func TestOpenAIGatewayService_SelectAccountWithScheduler_FailsOpenWhenAllProxiesQuarantined(t *testing.T) { + resetOpenAIAdvancedSchedulerSettingCacheForTest() + + // Every schedulable account shares the quarantined proxy: the circuit must + // degrade to a preference instead of zeroing out capacity (#5056). + proxyA := int64(5056) + accounts := []Account{ + {ID: 505601, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0, ProxyID: &proxyA}, + {ID: 505602, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5, ProxyID: &proxyA}, + } + 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, + }), + } + tripped, _ := svc.openaiProxyStreamCircuit.recordFailure(proxyA, time.Now()) + require.True(t, tripped) + + selection, _, err := svc.SelectAccountWithScheduler( + context.Background(), nil, "", "", "gpt-5.6-sol", nil, OpenAIUpstreamTransportAny, false, + ) + require.NoError(t, err, "quarantine must fail open instead of returning no available accounts") + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + require.NotNil(t, selection.Account.ProxyID) + require.Equal(t, proxyA, *selection.Account.ProxyID) + require.True(t, svc.openaiProxyStreamCircuit.isBlocked(proxyA, time.Now()), + "fail-open must not clear the quarantine; only a completed stream or TTL expiry does") +} + func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyRateLimitedAccountFallsBackToFreshCandidate(t *testing.T) { ctx := context.Background() groupID := int64(10101) diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index 01316cb58a..7c0ca8f033 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -1243,7 +1243,7 @@ func (s *OpenAIGatewayService) resolveFreshSchedulableOpenAIAccount(ctx context. if s.isOpenAIAccountRequestRuntimeBlocked(fresh, requestedModel) { return nil } - if s.isOpenAIProxyStreamQuarantined(fresh) { + if s.isOpenAIProxyStreamQuarantined(ctx, fresh) { return nil } return fresh @@ -1275,7 +1275,7 @@ func (s *OpenAIGatewayService) recheckSelectedOpenAIAccountFromDB(ctx context.Co if !parentHealthyForShadow(account, s.parentAccountLookup(ctx)) { return nil } - if s.isOpenAIProxyStreamQuarantined(account) { + if s.isOpenAIProxyStreamQuarantined(ctx, account) { return nil } return account @@ -1297,7 +1297,7 @@ func (s *OpenAIGatewayService) recheckSelectedOpenAIAccountFromDB(ctx context.Co if s.isOpenAIAccountRequestRuntimeBlocked(latest, requestedModel) { return nil } - if s.isOpenAIProxyStreamQuarantined(latest) { + if s.isOpenAIProxyStreamQuarantined(ctx, latest) { return nil } return latest diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index c06a989908..78b8e8f995 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -414,20 +414,21 @@ type OpenAIGatewayService struct { liveAttestation liveattestation.Provider liveAttestationCipher SecretEncryptor - openaiWSPoolOnce sync.Once - openaiWSStateStoreOnce sync.Once - openaiSchedulerOnce sync.Once - openaiProxyStreamCircuitOnce sync.Once - openaiWSPassthroughDialerOnce sync.Once - openaiModelTransientOnce sync.Once - agentIdentityTaskMu sync.Mutex - openaiWSPool *openAIWSConnPool - openaiWSStateStore OpenAIWSStateStore - openaiScheduler OpenAIAccountScheduler - openaiWSPassthroughDialer openAIWSClientDialer - openaiAccountStats *openAIAccountRuntimeStats - openaiModelTransient *openAIAccountModelTransientState - openaiProxyStreamCircuit *openAIProxyStreamCircuit + openaiWSPoolOnce sync.Once + openaiWSStateStoreOnce sync.Once + openaiSchedulerOnce sync.Once + openaiProxyStreamCircuitOnce sync.Once + openaiWSPassthroughDialerOnce sync.Once + openaiModelTransientOnce sync.Once + agentIdentityTaskMu sync.Mutex + openaiWSPool *openAIWSConnPool + openaiWSStateStore OpenAIWSStateStore + openaiScheduler OpenAIAccountScheduler + openaiWSPassthroughDialer openAIWSClientDialer + openaiAccountStats *openAIAccountRuntimeStats + openaiModelTransient *openAIAccountModelTransientState + openaiProxyStreamCircuit *openAIProxyStreamCircuit + openaiProxyStreamFailOpenLogAt atomic.Int64 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 d8cfba73b8..c23f74b528 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -1356,6 +1356,14 @@ func TestOpenAIStreamingPostOutputDisconnectQuarantinesSharedProxyWithoutSameStr svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{ MaxLineSize: defaultMaxLineSize, }}} + // collapseInterval 0: the two loop iterations below record within the + // production collapse window and must count as distinct failure events here. + svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 2, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 16, + }) for _, readErr := range []error{ io.ErrUnexpectedEOF, @@ -2008,6 +2016,14 @@ func TestOpenAIStreamingPassthroughPostOutputDisconnectQuarantinesSharedProxy(t proxyID := int64(4698) account := &Account{ID: 469804, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, ProxyID: &proxyID} svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}}} + // collapseInterval 0: the loop below records within the production collapse + // window and must count as distinct failure events here. + svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 2, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 16, + }) for _, readErr := range []error{io.ErrUnexpectedEOF, errors.New("http2: client connection lost")} { rec := httptest.NewRecorder() @@ -2029,7 +2045,7 @@ func TestOpenAIStreamingPassthroughPostOutputDisconnectQuarantinesSharedProxy(t require.Contains(t, rec.Body.String(), "partial") } - require.True(t, svc.isOpenAIProxyStreamQuarantined(account)) + require.True(t, svc.isOpenAIProxyStreamQuarantined(context.Background(), account)) } func TestOpenAIStreamingPassthroughResponseFailedBeforeOutputReturnsFailover(t *testing.T) { diff --git a/backend/internal/service/openai_proxy_stream_circuit.go b/backend/internal/service/openai_proxy_stream_circuit.go index a28a005143..575fc134bb 100644 --- a/backend/internal/service/openai_proxy_stream_circuit.go +++ b/backend/internal/service/openai_proxy_stream_circuit.go @@ -15,20 +15,32 @@ const ( defaultOpenAIProxyStreamFailureWindow = time.Minute defaultOpenAIProxyStreamQuarantineTTL = 10 * time.Minute defaultOpenAIProxyStreamCircuitMaxEntries = 4096 + // defaultOpenAIProxyStreamFailureCollapse merges disconnects that land + // within this interval into a single failure event. One proxy or HTTP/2 + // connection loss kills every stream multiplexed on it at the same + // moment; counting each stream separately would trip the breaker from a + // single upstream event (#5056). + defaultOpenAIProxyStreamFailureCollapse = 3 * time.Second + // openAIProxyStreamFailOpenLogInterval rate-limits the fail-open warning + // so an outage-mode burst does not flood the log. + openAIProxyStreamFailOpenLogInterval = 5 * time.Second ) type openAIProxyStreamCircuitSettings struct { + disabled bool failureThreshold int failureWindow time.Duration quarantineTTL time.Duration + collapseInterval time.Duration maxEntries int } type openAIProxyStreamCircuitEntry struct { - failureCount int - windowStart time.Time - blockedUntil time.Time - lastTouched time.Time + failureCount int + windowStart time.Time + lastFailureAt time.Time + blockedUntil time.Time + lastTouched time.Time } // openAIProxyStreamCircuit is an in-process, proxy-ID keyed circuit. It is @@ -45,12 +57,14 @@ func resolveOpenAIProxyStreamCircuitSettings(s *OpenAIGatewayService) openAIProx failureThreshold: defaultOpenAIProxyStreamFailureThreshold, failureWindow: defaultOpenAIProxyStreamFailureWindow, quarantineTTL: defaultOpenAIProxyStreamQuarantineTTL, + collapseInterval: defaultOpenAIProxyStreamFailureCollapse, maxEntries: defaultOpenAIProxyStreamCircuitMaxEntries, } if s == nil || s.cfg == nil { return settings } cfg := s.cfg.Gateway.OpenAIProxyStreamCircuit + settings.disabled = cfg.Disabled if cfg.FailureThreshold > 0 { settings.failureThreshold = cfg.FailureThreshold } @@ -76,6 +90,9 @@ func newOpenAIProxyStreamCircuit(settings openAIProxyStreamCircuitSettings) *ope if settings.maxEntries <= 0 { settings.maxEntries = defaultOpenAIProxyStreamCircuitMaxEntries } + if settings.collapseInterval < 0 { + settings.collapseInterval = 0 + } return &openAIProxyStreamCircuit{ settings: settings, entries: make(map[int64]openAIProxyStreamCircuitEntry), @@ -95,7 +112,7 @@ func (s *OpenAIGatewayService) getOpenAIProxyStreamCircuit() *openAIProxyStreamC } func (c *openAIProxyStreamCircuit) recordFailure(proxyID int64, now time.Time) (bool, time.Time) { - if c == nil || proxyID <= 0 { + if c == nil || c.settings.disabled || proxyID <= 0 { return false, time.Time{} } c.mu.Lock() @@ -115,7 +132,17 @@ func (c *openAIProxyStreamCircuit) recordFailure(proxyID int64, now time.Time) ( entry.windowStart = now entry.blockedUntil = time.Time{} } + // Collapse a burst of disconnects into one failure event: when a proxy or + // a multiplexed HTTP/2 connection dies, every in-flight stream reports the + // same underlying incident within moments of each other. + if c.settings.collapseInterval > 0 && !entry.lastFailureAt.IsZero() && + now.Sub(entry.lastFailureAt) >= 0 && now.Sub(entry.lastFailureAt) < c.settings.collapseInterval { + entry.lastTouched = now + c.entries[proxyID] = entry + return false, time.Time{} + } entry.failureCount++ + entry.lastFailureAt = now entry.lastTouched = now tripped := entry.failureCount >= c.settings.failureThreshold if tripped { @@ -139,7 +166,7 @@ func (c *openAIProxyStreamCircuit) recordSuccess(proxyID int64) bool { } func (c *openAIProxyStreamCircuit) isBlocked(proxyID int64, now time.Time) bool { - if c == nil || proxyID <= 0 { + if c == nil || c.settings.disabled || proxyID <= 0 { return false } c.mu.Lock() @@ -155,6 +182,25 @@ func (c *openAIProxyStreamCircuit) isBlocked(proxyID int64, now time.Time) bool return true } +// activeBlockCount reports how many proxies are currently quarantined. It +// gates the fail-open retry: a "no available accounts" selection result only +// warrants a second, quarantine-blind pass when the circuit is actually +// withholding capacity. +func (c *openAIProxyStreamCircuit) activeBlockCount(now time.Time) int { + if c == nil || c.settings.disabled { + return 0 + } + c.mu.Lock() + defer c.mu.Unlock() + count := 0 + for _, entry := range c.entries { + if !entry.blockedUntil.IsZero() && now.Before(entry.blockedUntil) { + count++ + } + } + return count +} + func (c *openAIProxyStreamCircuit) ensureCapacityLocked(now time.Time) { if len(c.entries) < c.settings.maxEntries { return @@ -219,11 +265,48 @@ func (s *OpenAIGatewayService) clearOpenAIProxyStreamDisconnect(account *Account } } -func (s *OpenAIGatewayService) isOpenAIProxyStreamQuarantined(account *Account) bool { +// openAIProxyStreamQuarantineBypassKey marks a selection pass that must ignore +// proxy quarantine. It is set for the second, fail-open selection attempt when +// the first pass found no available account while the circuit was withholding +// proxies: a degraded proxy is strictly better than answering 502 (#5056). +type openAIProxyStreamQuarantineBypassKey struct{} + +func withOpenAIProxyStreamQuarantineBypass(ctx context.Context) context.Context { + return context.WithValue(ctx, openAIProxyStreamQuarantineBypassKey{}, true) +} + +func openAIProxyStreamQuarantineBypassed(ctx context.Context) bool { + if ctx == nil { + return false + } + bypassed, _ := ctx.Value(openAIProxyStreamQuarantineBypassKey{}).(bool) + return bypassed +} + +func (s *OpenAIGatewayService) isOpenAIProxyStreamQuarantined(ctx context.Context, account *Account) bool { proxyID, ok := openAIProxyStreamCircuitProxyID(account) if !ok { return false } + if openAIProxyStreamQuarantineBypassed(ctx) { + return false + } circuit := s.getOpenAIProxyStreamCircuit() return circuit != nil && circuit.isBlocked(proxyID, time.Now()) } + +// logOpenAIProxyStreamQuarantineFailOpen emits a rate-limited warning when a +// selection pass had to re-admit quarantined proxies to serve at all. +func (s *OpenAIGatewayService) logOpenAIProxyStreamQuarantineFailOpen(requestedModel string, blockedProxies int) { + now := time.Now().UnixNano() + last := s.openaiProxyStreamFailOpenLogAt.Load() + if now-last < int64(openAIProxyStreamFailOpenLogInterval) || + !s.openaiProxyStreamFailOpenLogAt.CompareAndSwap(last, now) { + return + } + logger.L().With(zap.String("component", "service.openai_gateway")).Warn( + "openai.proxy_stream_quarantine_fail_open", + zap.Int("blocked_proxies", blockedProxies), + zap.String("model", requestedModel), + ) +} diff --git a/backend/internal/service/openai_proxy_stream_circuit_test.go b/backend/internal/service/openai_proxy_stream_circuit_test.go index c278ea8c1e..76a07f9eae 100644 --- a/backend/internal/service/openai_proxy_stream_circuit_test.go +++ b/backend/internal/service/openai_proxy_stream_circuit_test.go @@ -1,6 +1,7 @@ package service import ( + "context" "testing" "time" @@ -35,6 +36,82 @@ func TestOpenAIProxyStreamCircuitThresholdTTLAndSuccessReset(t *testing.T) { require.False(t, tripped, "failures outside the window must not accumulate") } +func TestOpenAIProxyStreamCircuitCollapsesBurstFailures(t *testing.T) { + base := time.Unix(1_800_000_000, 0) + circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 2, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + collapseInterval: 3 * time.Second, + maxEntries: 16, + }) + + // One HTTP/2 connection loss kills several multiplexed streams at once: + // the near-simultaneous reports must count as a single failure event. + tripped, _ := circuit.recordFailure(1, base) + require.False(t, tripped) + tripped, _ = circuit.recordFailure(1, base.Add(time.Second)) + require.False(t, tripped, "burst failures inside the collapse interval must merge") + tripped, _ = circuit.recordFailure(1, base.Add(2*time.Second)) + require.False(t, tripped, "burst failures inside the collapse interval must merge") + require.False(t, circuit.isBlocked(1, base.Add(2*time.Second))) + + // A second, distinct incident past the collapse interval still trips. + tripped, _ = circuit.recordFailure(1, base.Add(5*time.Second)) + require.True(t, tripped) + require.True(t, circuit.isBlocked(1, base.Add(5*time.Second))) +} + +func TestOpenAIProxyStreamCircuitDisabled(t *testing.T) { + base := time.Unix(1_800_000_000, 0) + circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + disabled: true, + failureThreshold: 1, + 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.Equal(t, 0, circuit.activeBlockCount(base)) +} + +func TestOpenAIProxyStreamCircuitActiveBlockCount(t *testing.T) { + base := time.Unix(1_800_000_000, 0) + circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 1, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 16, + }) + + require.Equal(t, 0, circuit.activeBlockCount(base)) + tripped, until := circuit.recordFailure(1, base) + require.True(t, tripped) + circuit.recordFailure(2, base) // second proxy also tripped (threshold 1) + require.Equal(t, 2, circuit.activeBlockCount(base.Add(time.Second))) + require.Equal(t, 0, circuit.activeBlockCount(until), "expired quarantines must not count") +} + +func TestOpenAIProxyStreamQuarantineBypassContext(t *testing.T) { + proxyID := int64(7) + account := &Account{ID: 1, Platform: PlatformOpenAI, ProxyID: &proxyID} + svc := &OpenAIGatewayService{} + svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 1, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 16, + }) + svc.openaiProxyStreamCircuit.recordFailure(proxyID, time.Now()) + + ctx := context.Background() + require.True(t, svc.isOpenAIProxyStreamQuarantined(ctx, account)) + require.False(t, svc.isOpenAIProxyStreamQuarantined(withOpenAIProxyStreamQuarantineBypass(ctx), account)) +} + func TestOpenAIProxyStreamCircuitBoundsEntries(t *testing.T) { base := time.Unix(1_800_000_000, 0) circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index e030ae3649..9b62151db6 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -423,7 +423,10 @@ gateway: fallback_window_seconds: 60 fallback_ttl_seconds: 600 # OpenAI Responses SSE 代理断流熔断；按 proxy_id 跨账号隔离，仅影响下一次调度。 + # 隔离是偏好而非硬闸：当全部候选账号都在被隔离代理下时自动放行（fail-open）， + # 不会因熔断而返回"无可用账号"。3 秒内的并发断流合并为一次失败事件。 openai_proxy_stream_circuit: + disabled: false failure_threshold: 2 window_seconds: 60 ttl_seconds: 600