diff --git a/backend/internal/repository/http_upstream.go b/backend/internal/repository/http_upstream.go index bdd6478cff5b..286fac12e6bc 100644 --- a/backend/internal/repository/http_upstream.go +++ b/backend/internal/repository/http_upstream.go @@ -136,6 +136,7 @@ type upstreamClientEntry struct { type openAIHTTP2FallbackState struct { mu sync.Mutex windowStart time.Time + lastFailureAt time.Time errorCount int fallbackUntil time.Time } @@ -213,6 +214,7 @@ func (s *httpUpstreamService) Do(req *http.Request, proxyURL string, accountID i } // 执行请求 + requestStartedAt := time.Now() client := httpClientForUpstreamRequest(entry.client, req) client = httpClientWithGrokAccessDeniedFallback(client) resp, err := servertiming.Do(client, req) @@ -223,17 +225,23 @@ func (s *httpUpstreamService) Do(req *http.Request, proxyURL string, accountID i atomic.StoreInt64(&entry.lastUsed, time.Now().UnixNano()) return nil, err } - s.recordOpenAIHTTP2Success(profile, entry.protocolMode, entry.proxyKey) // 如果上游返回了压缩内容，解压后再交给业务层 decompressResponseBody(resp) - // 包装响应体，在关闭时自动减少计数并更新时间戳 + // H2 streams can fail after a successful response header. Classify the whole + // body lifecycle so repeated mid-stream failures can activate the existing + // per-proxy H2 -> H1 compatibility fallback instead of being reset by 200 headers. // 这确保了流式响应（如 SSE）在完全读取前不会被淘汰 - resp.Body = wrapTrackedBody(resp.Body, func() { - atomic.AddInt64(&entry.inFlight, -1) - atomic.StoreInt64(&entry.lastUsed, time.Now().UnixNano()) - }) + resp.Body = wrapTrackedBodyWithOutcome( + resp.Body, + func() { + atomic.AddInt64(&entry.inFlight, -1) + atomic.StoreInt64(&entry.lastUsed, time.Now().UnixNano()) + }, + func() { s.recordOpenAIHTTP2Success(profile, entry.protocolMode, entry.proxyKey, requestStartedAt) }, + func(err error) { s.recordOpenAIHTTP2Failure(profile, entry.protocolMode, entry.proxyKey, err) }, + ) return resp, nil } @@ -1057,6 +1065,7 @@ func isOpenAIHTTP2CompatibilityError(err error) bool { "no application protocol", "protocol error", "stream error", + "http2: client connection lost", "goaway", "refused_stream", "frame too large", @@ -1119,7 +1128,12 @@ func (s *httpUpstreamService) recordOpenAIHTTP2Failure(profile service.HTTPUpstr } } -func (s *httpUpstreamService) recordOpenAIHTTP2Success(profile service.HTTPUpstreamProfile, protocolMode, proxyKey string) { +func (s *httpUpstreamService) recordOpenAIHTTP2Success( + profile service.HTTPUpstreamProfile, + protocolMode string, + proxyKey string, + requestStartedAt time.Time, +) { if profile != service.HTTPUpstreamProfileOpenAI || protocolMode != upstreamProtocolModeOpenAIH2 { return } @@ -1134,7 +1148,7 @@ func (s *httpUpstreamService) recordOpenAIHTTP2Success(profile service.HTTPUpstr if !ok || state == nil { return } - state.resetErrorWindow() + state.resetErrorWindow(requestStartedAt) } func (s *openAIHTTP2FallbackState) isFallbackActive(now time.Time) bool { @@ -1150,10 +1164,14 @@ func (s *openAIHTTP2FallbackState) isFallbackActive(now time.Time) bool { return false } -func (s *openAIHTTP2FallbackState) resetErrorWindow() { +func (s *openAIHTTP2FallbackState) resetErrorWindow(requestStartedAt time.Time) { s.mu.Lock() defer s.mu.Unlock() + if !s.lastFailureAt.IsZero() && requestStartedAt.Before(s.lastFailureAt) { + return + } s.windowStart = time.Time{} + s.lastFailureAt = time.Time{} s.errorCount = 0 } @@ -1183,6 +1201,7 @@ func (s *openAIHTTP2FallbackState) recordFailure(now time.Time, threshold int, w s.errorCount = 0 } s.errorCount++ + s.lastFailureAt = now if s.errorCount < threshold { return false, time.Time{} } @@ -1422,18 +1441,49 @@ type trackedBody struct { io.ReadCloser // 原始响应体 once sync.Once onClose func() // 关闭时的回调函数 + outcomeOnce sync.Once + onSuccess func() + onReadError func(error) +} + +func (b *trackedBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + if err == nil { + return n, nil + } + if errors.Is(err, io.EOF) { + b.recordSuccess() + } else { + b.recordReadError(err) + } + return n, err } // Close 关闭响应体并执行回调 // 使用 sync.Once 确保回调只执行一次 func (b *trackedBody) Close() error { err := b.ReadCloser.Close() + if err != nil { + b.recordReadError(err) + } if b.onClose != nil { b.once.Do(b.onClose) } return err } +func (b *trackedBody) recordSuccess() { + if b.onSuccess != nil { + b.outcomeOnce.Do(b.onSuccess) + } +} + +func (b *trackedBody) recordReadError(err error) { + if b.onReadError != nil { + b.outcomeOnce.Do(func() { b.onReadError(err) }) + } +} + // wrapTrackedBody 包装响应体以跟踪关闭事件 // 用于在响应体关闭时更新 inFlight 计数 // @@ -1444,10 +1494,24 @@ func (b *trackedBody) Close() error { // 返回: // - io.ReadCloser: 包装后的响应体 func wrapTrackedBody(body io.ReadCloser, onClose func()) io.ReadCloser { + return wrapTrackedBodyWithOutcome(body, onClose, nil, nil) +} + +func wrapTrackedBodyWithOutcome( + body io.ReadCloser, + onClose func(), + onSuccess func(), + onReadError func(error), +) io.ReadCloser { if body == nil { return body } - return &trackedBody{ReadCloser: body, onClose: onClose} + return &trackedBody{ + ReadCloser: body, + onClose: onClose, + onSuccess: onSuccess, + onReadError: onReadError, + } } // decompressResponseBody 根据 Content-Encoding 解压响应体。 diff --git a/backend/internal/repository/http_upstream_test.go b/backend/internal/repository/http_upstream_test.go index 3c7840dc47d8..dfb1421ab325 100644 --- a/backend/internal/repository/http_upstream_test.go +++ b/backend/internal/repository/http_upstream_test.go @@ -731,6 +731,134 @@ func (s *HTTPUpstreamSuite) TestOpenAIHTTP2ProxyCompatibilityErrorActivatesFallb require.Equal(s.T(), upstreamProtocolModeOpenAIH1Fallback, entry.protocolMode) } +func (s *HTTPUpstreamSuite) TestOpenAIHTTP2BodyFailuresActivateProxyFallback() { + s.cfg.Gateway = config.GatewayConfig{ + OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{ + Enabled: true, + AllowProxyFallbackToHTTP1: true, + FallbackErrorThreshold: 2, + FallbackWindowSeconds: 60, + FallbackTTLSeconds: 600, + }, + } + svc := s.newService() + proxyURL := "http://proxy.local:8080" + readBody := func(readErr error) { + startedAt := time.Now() + body := wrapTrackedBodyWithOutcome( + io.NopCloser(&failingReader{err: readErr}), + nil, + func() { + svc.recordOpenAIHTTP2Success(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, startedAt) + }, + func(err error) { + svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, err) + }, + ) + _, err := io.ReadAll(body) + require.Error(s.T(), err) + require.NoError(s.T(), body.Close()) + } + + readBody(errors.New("http2: client connection lost")) + require.False(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL)) + readBody(errors.New("http2: client connection lost")) + require.True(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL)) +} + +func (s *HTTPUpstreamSuite) TestOpenAIHTTP2GenericUnexpectedEOFDoesNotActivateProxyFallback() { + s.cfg.Gateway = config.GatewayConfig{ + OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{ + Enabled: true, + AllowProxyFallbackToHTTP1: true, + FallbackErrorThreshold: 1, + FallbackWindowSeconds: 60, + FallbackTTLSeconds: 600, + }, + } + svc := s.newService() + proxyURL := "http://proxy.local:8080" + body := wrapTrackedBodyWithOutcome( + io.NopCloser(&failingReader{err: io.ErrUnexpectedEOF}), + nil, + nil, + func(err error) { + svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, err) + }, + ) + + _, err := io.ReadAll(body) + require.ErrorIs(s.T(), err, io.ErrUnexpectedEOF) + require.NoError(s.T(), body.Close()) + require.False(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL)) +} + +func (s *HTTPUpstreamSuite) TestOpenAIHTTP2SuccessfulBodyResetsFailureWindow() { + s.cfg.Gateway = config.GatewayConfig{ + OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{ + Enabled: true, + AllowProxyFallbackToHTTP1: true, + FallbackErrorThreshold: 2, + FallbackWindowSeconds: 60, + FallbackTTLSeconds: 600, + }, + } + svc := s.newService() + proxyURL := "http://proxy.local:8080" + failed := func() { + startedAt := time.Now() + body := wrapTrackedBodyWithOutcome( + io.NopCloser(&failingReader{err: errors.New("http2: client connection lost")}), + nil, + func() { + svc.recordOpenAIHTTP2Success(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, startedAt) + }, + func(err error) { + svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, err) + }, + ) + _, err := io.ReadAll(body) + require.Error(s.T(), err) + require.NoError(s.T(), body.Close()) + } + succeeded := func() { + startedAt := time.Now() + body := wrapTrackedBodyWithOutcome( + io.NopCloser(strings.NewReader("ok")), + nil, + func() { + svc.recordOpenAIHTTP2Success(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, startedAt) + }, + func(err error) { + svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, err) + }, + ) + _, err := io.ReadAll(body) + require.NoError(s.T(), err) + require.NoError(s.T(), body.Close()) + } + + failed() + succeeded() + failed() + require.False(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL)) +} + +func (s *HTTPUpstreamSuite) TestOpenAIHTTP2OlderSuccessCannotResetNewFailure() { + base := time.Unix(1_800_000_000, 0) + state := &openAIHTTP2FallbackState{} + + tripped, _ := state.recordFailure(base, 2, time.Minute, 10*time.Minute) + require.False(s.T(), tripped) + state.resetErrorWindow(base.Add(-time.Second)) + tripped, _ = state.recordFailure(base.Add(10*time.Second), 2, time.Minute, 10*time.Minute) + require.True(s.T(), tripped, "a success from an older request must not erase a newer transport failure") +} + +type failingReader struct{ err error } + +func (r *failingReader) Read([]byte) (int, error) { return 0, r.err } + // TestNormalizeProxyURL_Canonicalizes 测试代理 URL 规范化 // 验证等价地址能够映射到同一缓存键 func (s *HTTPUpstreamSuite) TestNormalizeProxyURL_Canonicalizes() { diff --git a/backend/internal/service/openai_proxy_stream_circuit.go b/backend/internal/service/openai_proxy_stream_circuit.go index 575fc134bb76..a44f8194f776 100644 --- a/backend/internal/service/openai_proxy_stream_circuit.go +++ b/backend/internal/service/openai_proxy_stream_circuit.go @@ -158,7 +158,11 @@ func (c *openAIProxyStreamCircuit) recordSuccess(proxyID int64) bool { } c.mu.Lock() defer c.mu.Unlock() - if _, ok := c.entries[proxyID]; !ok { + entry, ok := c.entries[proxyID] + if !ok || !entry.blockedUntil.IsZero() { + // A success can belong to a request that started before a concurrent + // failure tripped the circuit. Once quarantined, keep the proxy blocked + // for the configured TTL; fail-open selection still preserves capacity. return false } delete(c.entries, proxyID) diff --git a/backend/internal/service/openai_proxy_stream_circuit_test.go b/backend/internal/service/openai_proxy_stream_circuit_test.go index 76a07f9eae4b..20777c39e2ea 100644 --- a/backend/internal/service/openai_proxy_stream_circuit_test.go +++ b/backend/internal/service/openai_proxy_stream_circuit_test.go @@ -28,6 +28,8 @@ func TestOpenAIProxyStreamCircuitThresholdTTLAndSuccessReset(t *testing.T) { 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.recordSuccess(1), "an older in-flight success must not clear an active quarantine") + 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)