seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-14 12:02:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #6281 from wuji-labs/feat/http2-long-stream-keepalive

Browse Source 
feat(upstream): add HTTP/2 PING keepalive for long LLM generation streams
This commit is contained in:
Wesley Liddick
2026-09-08 20:47:07 +08:00
committed by GitHub
parent f9d7826711 f5e4da5e72
commit 233f3dffe6
5 changed files with 63 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/http_upstream.go
+14 -10
View File
@@ -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 时不可淘汰
}
@@ -1007,6 +1008,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
}
@@ -1332,11 +1336,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:
@@ -1353,18 +1357,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
}
backend/internal/repository/http_upstream_http2_keepalive_test.go
+19 -19
View File
@@ -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 生效")
}
backend/internal/repository/http_upstream_test.go
+18
View File
@@ -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.True(s.T(), transport.Protocols.HTTP2(), "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,
backend/internal/service/http_upstream_profile.go
+5 -4
View File
@@ -7,9 +7,10 @@ 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"
)
type httpUpstreamProfileContextKey struct{}
@@ -37,7 +38,7 @@ func HTTPUpstreamProfileFromContext(ctx context.Context) HTTPUpstreamProfile {
return HTTPUpstreamProfileDefault
}
switch profile {
case HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileGrok:
case HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileGrok, HTTPUpstreamProfileLongStream:
return profile
default:
return HTTPUpstreamProfileDefault
backend/internal/service/http_upstream_profile_test.go
+7
View File
@@ -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)