From 81fd8530056c6480df8e69770a85d8f411340290 Mon Sep 17 00:00:00 2001 From: WUJI-Labs Date: Thu, 27 Aug 2026 15:55:05 +0800 Subject: [PATCH 1/3] feat(upstream): add HTTP/2 PING keepalive for long LLM generation streams Long-running LLM inferences (such as high-thinking Gemini models) may experience silent TCP disconnections across intermediate proxies, NAT gateways, or firewalls when no data frames are emitted for tens of seconds to minutes. Go's default http2.Transport sets ReadIdleTimeout=0 (disabling active PING checks). This commit introduces HTTPUpstreamProfileLongStream with: - ReadIdleTimeout = 10s - PingTimeout = 5s When a pooled H2 connection stays idle for 10s during long streams, active PING frames are dispatched. Dead connections failing to respond within 5s are proactively evicted, preventing client generation requests from hanging on dead sockets or failing with EOF/502. Signed-off-by: WUJI-Labs --- backend/internal/repository/http_upstream.go | 24 +++++++++++-------- .../internal/repository/http_upstream_test.go | 18 ++++++++++++++ .../internal/service/http_upstream_profile.go | 3 ++- .../service/http_upstream_profile_test.go | 7 ++++++ 4 files changed, 41 insertions(+), 11 deletions(-) diff --git a/backend/internal/repository/http_upstream.go b/backend/internal/repository/http_upstream.go index bdd6478cff..6b611c4a30 100644 --- a/backend/internal/repository/http_upstream.go +++ b/backend/internal/repository/http_upstream.go @@ -76,13 +76,13 @@ const ( defaultOpenAIHTTP2FallbackErrorThreshold = 2 defaultOpenAIHTTP2FallbackWindow = 60 * time.Second defaultOpenAIHTTP2FallbackTTL = 10 * time.Minute - // OpenAI HTTP/2 连接健康探测：Codex 上游改走 HTTP/2 后，池化连接被代理/NAT + // 长流 HTTP/2 连接健康探测：池化连接被代理/NAT // 静默掐断会成为“死连接”（两端都以为存活），请求落上去会挂到 TCP 重传超时 // （分钟级）。Go 的 http2.Transport 默认 ReadIdleTimeout=0（不发健康 PING）， // 无法检测。启用主动 PING 探测：连接空闲 ReadIdleTimeout 后发 PING，PingTimeout // 内无响应即判定死连接并关闭，从源头避免请求挂在死连接上。 - openAIHTTP2ReadIdleTimeout = 15 * time.Second - openAIHTTP2PingTimeout = 15 * time.Second + longStreamHTTP2ReadIdleTimeout = 10 * time.Second + longStreamHTTP2PingTimeout = 5 * time.Second // The Grok CLI proxy rejects requests that do not identify a supported // client version. Host/env/version pins live in package xai so service, @@ -96,6 +96,7 @@ const ( const ( upstreamProtocolModeDefault = "default" + upstreamProtocolModeLongStreamH2 = "long_stream_h2" upstreamProtocolModeOpenAIH1 = "openai_h1" upstreamProtocolModeOpenAIH2 = "openai_h2" upstreamProtocolModeOpenAIH1Fallback = "openai_h1_fallback" @@ -128,7 +129,7 @@ type upstreamClientEntry struct { client *http.Client // HTTP 客户端实例 proxyKey string // 代理标识（用于检测代理变更） poolKey string // 连接池配置标识（用于检测配置变更） - protocolMode string // 协议模式（default/openai_h1/openai_h2/openai_h1_fallback） + protocolMode string // 协议模式（default/long_stream_h2/openai_h1/openai_h2/openai_h1_fallback） lastUsed int64 // 最后使用时间戳（纳秒），用于 LRU 淘汰 inFlight int64 // 当前进行中的请求数，>0 时不可淘汰 } @@ -992,6 +993,9 @@ func (s *httpUpstreamService) resolveOpenAIHTTP2Settings() openAIHTTP2Settings { } func (s *httpUpstreamService) resolveProtocolMode(profile service.HTTPUpstreamProfile, proxyKey string, parsedProxy *url.URL) string { + if profile == service.HTTPUpstreamProfileLongStream { + return upstreamProtocolModeLongStreamH2 + } if profile == service.HTTPUpstreamProfileGrok { return upstreamProtocolModeGrok } @@ -1317,11 +1321,11 @@ func buildUpstreamTransport(settings poolSettings, proxyURL *url.URL, protocolMo ResponseHeaderTimeout: settings.responseHeaderTimeout, } switch protocolMode { - case upstreamProtocolModeOpenAIH2: + case upstreamProtocolModeLongStreamH2, upstreamProtocolModeOpenAIH2: transport.ForceAttemptHTTP2 = true // 显式配置 http2 并启用 PING 健康探测，剔除代理/NAT 静默掐断的死连接， // 避免请求挂在死连接上直到 TCP 重传超时（分钟级）。 - if _, err := enableOpenAIHTTP2KeepAlive(transport); err != nil { + if _, err := enableHTTP2KeepAlive(transport); err != nil { return nil, err } case upstreamProtocolModeOpenAIH1: @@ -1338,18 +1342,18 @@ func buildUpstreamTransport(settings poolSettings, proxyURL *url.URL, protocolMo return transport, nil } -// enableOpenAIHTTP2KeepAlive 在 http.Transport 上显式配置 HTTP/2 并启用连接健康探测。 +// enableHTTP2KeepAlive 在 http.Transport 上显式配置 HTTP/2 并启用连接健康探测。 // Go 默认惰性配置 http2 且 ReadIdleTimeout=0（不发健康 PING），无法检测被代理/NAT // 静默掐断的死连接。此处主动设置 ReadIdleTimeout/PingTimeout，让死连接被提前 PING // 出并关闭，请求得以重建连接而非挂到 TCP 重传超时。返回底层 *http2.Transport 便于测试。 -func enableOpenAIHTTP2KeepAlive(transport *http.Transport) (*http2.Transport, error) { +func enableHTTP2KeepAlive(transport *http.Transport) (*http2.Transport, error) { h2, err := http2.ConfigureTransports(transport) if err != nil { return nil, err } if h2 != nil { - h2.ReadIdleTimeout = openAIHTTP2ReadIdleTimeout - h2.PingTimeout = openAIHTTP2PingTimeout + h2.ReadIdleTimeout = longStreamHTTP2ReadIdleTimeout + h2.PingTimeout = longStreamHTTP2PingTimeout } return h2, nil } diff --git a/backend/internal/repository/http_upstream_test.go b/backend/internal/repository/http_upstream_test.go index 3c7840dc47..6e6550b6d7 100644 --- a/backend/internal/repository/http_upstream_test.go +++ b/backend/internal/repository/http_upstream_test.go @@ -629,6 +629,24 @@ func (s *HTTPUpstreamSuite) TestOpenAIProfileDefaultsToHTTP2AndNoHeaderTimeout() require.Equal(s.T(), upstreamProtocolModeOpenAIH2, entry.protocolMode) } +func (s *HTTPUpstreamSuite) TestLongStreamProfileUsesSharedHTTP2KeepAlive() { + s.cfg.Gateway = config.GatewayConfig{ + ResponseHeaderTimeout: 600, + OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{ + Enabled: false, + }, + } + svc := s.newService() + entry, err := svc.getClientEntry("", 1, 1, service.HTTPUpstreamProfileLongStream, false, false) + require.NoError(s.T(), err) + transport, ok := entry.client.Transport.(*http.Transport) + require.True(s.T(), ok, "expected *http.Transport") + require.Equal(s.T(), 600*time.Second, transport.ResponseHeaderTimeout, "long-stream profile should retain the generic header timeout") + require.True(s.T(), transport.ForceAttemptHTTP2, "long-stream profile must enable HTTP/2 independently of OpenAI settings") + require.NotNil(s.T(), transport.TLSNextProto["h2"], "long-stream profile must install HTTP/2 PING health checks") + require.Equal(s.T(), upstreamProtocolModeLongStreamH2, entry.protocolMode) +} + func (s *HTTPUpstreamSuite) TestOpenAIProfileCustomHeaderTimeout() { s.cfg.Gateway = config.GatewayConfig{ ResponseHeaderTimeout: 600, diff --git a/backend/internal/service/http_upstream_profile.go b/backend/internal/service/http_upstream_profile.go index b2de6002cf..b12362ffac 100644 --- a/backend/internal/service/http_upstream_profile.go +++ b/backend/internal/service/http_upstream_profile.go @@ -10,6 +10,7 @@ const ( HTTPUpstreamProfileDefault HTTPUpstreamProfile = "" HTTPUpstreamProfileOpenAI HTTPUpstreamProfile = "openai" HTTPUpstreamProfileGrok HTTPUpstreamProfile = "grok" + HTTPUpstreamProfileLongStream HTTPUpstreamProfile = "long_stream" ) type httpUpstreamProfileContextKey struct{} @@ -36,7 +37,7 @@ func HTTPUpstreamProfileFromContext(ctx context.Context) HTTPUpstreamProfile { return HTTPUpstreamProfileDefault } switch profile { - case HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileGrok: + case HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileGrok, HTTPUpstreamProfileLongStream: return profile default: return HTTPUpstreamProfileDefault diff --git a/backend/internal/service/http_upstream_profile_test.go b/backend/internal/service/http_upstream_profile_test.go index 9cd4bf4ff9..2d48f0c647 100644 --- a/backend/internal/service/http_upstream_profile_test.go +++ b/backend/internal/service/http_upstream_profile_test.go @@ -20,6 +20,13 @@ func TestWithHTTPUpstreamProfile_OpenAI(t *testing.T) { } } +func TestWithHTTPUpstreamProfile_LongStream(t *testing.T) { + ctx := WithHTTPUpstreamProfile(context.TODO(), HTTPUpstreamProfileLongStream) + if profile := HTTPUpstreamProfileFromContext(ctx); profile != HTTPUpstreamProfileLongStream { + t.Fatalf("expected profile %q, got %q", HTTPUpstreamProfileLongStream, profile) + } +} + func TestWithHTTPUpstreamRedirectsDisabled(t *testing.T) { //nolint:staticcheck // Exercises the defensive nil-context fallback. ctx := WithHTTPUpstreamRedirectsDisabled(nil) From 9dc4c40bff675cadd95e78a106c5325472ba3f8a Mon Sep 17 00:00:00 2001 From: WUJI-Labs Date: Wed, 2 Sep 2026 12:28:37 +0800 Subject: [PATCH 2/3] test(upstream): align HTTP/2 keepalive tests with long-stream profile API --- .../http_upstream_http2_keepalive_test.go | 38 +++++++++---------- .../internal/repository/http_upstream_test.go | 2 +- 2 files changed, 20 insertions(+), 20 deletions(-) diff --git a/backend/internal/repository/http_upstream_http2_keepalive_test.go b/backend/internal/repository/http_upstream_http2_keepalive_test.go index 66a30dab0e..29acf4b631 100644 --- a/backend/internal/repository/http_upstream_http2_keepalive_test.go +++ b/backend/internal/repository/http_upstream_http2_keepalive_test.go @@ -32,44 +32,44 @@ func requireHTTP2Configured(t *testing.T, tr *http.Transport, msg string) { require.True(t, tr.Protocols.HTTP2(), msg) } -// Codex/OpenAI 上游改走 HTTP/2 后，池化连接被代理/NAT 静默掐断会成为“死连接”： +// 长流 / OpenAI 上游改走 HTTP/2 后，池化连接被代理/NAT 静默掐断会成为“死连接”： // 两端都以为连接存活，请求落上去会挂到 TCP 重传超时（分钟级）才失败。Go 的 // http2.Transport 默认 ReadIdleTimeout=0（不发健康 PING），无法检测这种死连接。 // 必须显式启用主动 PING 探测，让死连接被提前剔除，而不是只靠 ResponseHeaderTimeout // 事后兜底。 -func TestEnableOpenAIHTTP2KeepAlive_EnablesPingHealthCheck(t *testing.T) { +func TestEnableHTTP2KeepAlive_EnablesPingHealthCheck(t *testing.T) { tr := &http.Transport{} - h2, err := enableOpenAIHTTP2KeepAlive(tr) + h2, err := enableHTTP2KeepAlive(tr) require.NoError(t, err) require.NotNil(t, h2, "必须返回已配置的 *http2.Transport") require.Positive(t, h2.ReadIdleTimeout, "必须启用空闲 PING 探测以剔除死连接") - require.Equal(t, openAIHTTP2ReadIdleTimeout, h2.ReadIdleTimeout) - require.Equal(t, openAIHTTP2PingTimeout, h2.PingTimeout, "PING 无响应必须有超时判定") + require.Equal(t, longStreamHTTP2ReadIdleTimeout, h2.ReadIdleTimeout) + require.Equal(t, longStreamHTTP2PingTimeout, h2.PingTimeout, "PING 无响应必须有超时判定") requireHTTP2Configured(t, tr, "http2 必须已挂到底层 http.Transport 上") } -// openai_h2 模式构建的 Transport 必须带上 H2 PING 健康探测，从源头剔除死连接。 -func TestBuildUpstreamTransport_OpenAIH2_EnablesPingHealthCheck(t *testing.T) { - tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, upstreamProtocolModeOpenAIH2) +// long_stream_h2 模式构建的 Transport 必须带上 H2 PING 健康探测，从源头剔除死连接。 +func TestBuildUpstreamTransport_LongStreamH2_EnablesPingHealthCheck(t *testing.T) { + tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, upstreamProtocolModeLongStreamH2) require.NoError(t, err) - require.True(t, tr.ForceAttemptHTTP2, "openai_h2 必须启用 HTTP/2") - requireHTTP2Configured(t, tr, "openai_h2 必须显式配置 http2 以启用 ReadIdleTimeout") + require.True(t, tr.ForceAttemptHTTP2, "long_stream_h2 必须启用 HTTP/2") + requireHTTP2Configured(t, tr, "long_stream_h2 必须显式配置 http2 以启用 ReadIdleTimeout") } // 非 H2 模式（default/h1）不应因本次改动被误配置：default 走 Go 自动 H2（惰性配置， // 构建时 Protocols/TLSNextProto 仍为空），h1 模式显式禁用 H2。避免波及 Claude/Gemini 热路径。 -func TestBuildUpstreamTransport_NonOpenAIH2_NotEagerlyConfigured(t *testing.T) { +func TestBuildUpstreamTransport_NonLongStreamH2_NotEagerlyConfigured(t *testing.T) { tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, upstreamProtocolModeDefault) require.NoError(t, err) require.Nil(t, tr.Protocols, "default 模式不应在构建期主动配置 http2 keepalive") require.Nil(t, tr.TLSNextProto["h2"], "default 模式不应在构建期主动配置 http2 keepalive") } -// openai_h2 模式构建的 Transport 必须真正以 HTTP/2 与上游通信，PING 健康探测才有载体： -// 自定义 DialContext 下 Go 不会自动启用 H2，全靠 enableOpenAIHTTP2KeepAlive 的显式配置。 -func TestBuildUpstreamTransport_OpenAIH2_NegotiatesHTTP2(t *testing.T) { +// long_stream_h2 模式构建的 Transport 必须真正以 HTTP/2 与上游通信，PING 健康探测才有载体： +// 自定义 DialContext 下 Go 不会自动启用 H2，全靠 enableHTTP2KeepAlive 的显式配置。 +func TestBuildUpstreamTransport_LongStreamH2_NegotiatesHTTP2(t *testing.T) { srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) @@ -77,7 +77,7 @@ func TestBuildUpstreamTransport_OpenAIH2_NegotiatesHTTP2(t *testing.T) { srv.StartTLS() defer srv.Close() - tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, upstreamProtocolModeOpenAIH2) + tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, upstreamProtocolModeLongStreamH2) require.NoError(t, err) defer tr.CloseIdleConnections() require.NotNil(t, tr.TLSClientConfig) @@ -91,18 +91,18 @@ func TestBuildUpstreamTransport_OpenAIH2_NegotiatesHTTP2(t *testing.T) { require.NoError(t, err) require.NoError(t, resp.Body.Close()) require.Equal(t, http.StatusOK, resp.StatusCode) - require.Equal(t, 2, resp.ProtoMajor, "openai_h2 必须协商到 HTTP/2") + require.Equal(t, 2, resp.ProtoMajor, "long_stream_h2 必须协商到 HTTP/2") } // 死连接在经 HTTP 代理（CONNECT 隧道）时最高发，这是带 proxy 账号的真实生产路径： // 显式 http2 配置须与 Transport.Proxy 同时正确生效，不能相互干扰。 -func TestBuildUpstreamTransport_OpenAIH2_WithHTTPProxy_EnablesKeepAlive(t *testing.T) { +func TestBuildUpstreamTransport_LongStreamH2_WithHTTPProxy_EnablesKeepAlive(t *testing.T) { proxyURL, err := url.Parse("http://127.0.0.1:8080") require.NoError(t, err) - tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), proxyURL, upstreamProtocolModeOpenAIH2) + tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), proxyURL, upstreamProtocolModeLongStreamH2) require.NoError(t, err) require.True(t, tr.ForceAttemptHTTP2) - requireHTTP2Configured(t, tr, "经代理的 openai_h2 也必须启用 http2 keepalive") + requireHTTP2Configured(t, tr, "经代理的 long_stream_h2 也必须启用 http2 keepalive") require.NotNil(t, tr.Proxy, "HTTP 代理仍须通过 Transport.Proxy 生效") } diff --git a/backend/internal/repository/http_upstream_test.go b/backend/internal/repository/http_upstream_test.go index 6e6550b6d7..00cff96e1e 100644 --- a/backend/internal/repository/http_upstream_test.go +++ b/backend/internal/repository/http_upstream_test.go @@ -643,7 +643,7 @@ func (s *HTTPUpstreamSuite) TestLongStreamProfileUsesSharedHTTP2KeepAlive() { require.True(s.T(), ok, "expected *http.Transport") require.Equal(s.T(), 600*time.Second, transport.ResponseHeaderTimeout, "long-stream profile should retain the generic header timeout") require.True(s.T(), transport.ForceAttemptHTTP2, "long-stream profile must enable HTTP/2 independently of OpenAI settings") - require.NotNil(s.T(), transport.TLSNextProto["h2"], "long-stream profile must install HTTP/2 PING health checks") + require.True(s.T(), transport.Protocols.HTTP2(), "long-stream profile must install HTTP/2 PING health checks") require.Equal(s.T(), upstreamProtocolModeLongStreamH2, entry.protocolMode) } From f5e4da5e72acd4fb026e0e97e7dfc61eae29735c Mon Sep 17 00:00:00 2001 From: WUJI-Labs Date: Wed, 2 Sep 2026 12:40:23 +0800 Subject: [PATCH 3/3] style(upstream): gofmt http_upstream_profile.go --- backend/internal/service/http_upstream_profile.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/backend/internal/service/http_upstream_profile.go b/backend/internal/service/http_upstream_profile.go index b12362ffac..88dd117366 100644 --- a/backend/internal/service/http_upstream_profile.go +++ b/backend/internal/service/http_upstream_profile.go @@ -7,9 +7,9 @@ import "context" type HTTPUpstreamProfile string const ( - HTTPUpstreamProfileDefault HTTPUpstreamProfile = "" - HTTPUpstreamProfileOpenAI HTTPUpstreamProfile = "openai" - HTTPUpstreamProfileGrok HTTPUpstreamProfile = "grok" + HTTPUpstreamProfileDefault HTTPUpstreamProfile = "" + HTTPUpstreamProfileOpenAI HTTPUpstreamProfile = "openai" + HTTPUpstreamProfileGrok HTTPUpstreamProfile = "grok" HTTPUpstreamProfileLongStream HTTPUpstreamProfile = "long_stream" )