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

fix(upstream): 恢复 OpenAI HTTP/2 保活容错期限

Browse Source
This commit is contained in:
shaw
2026-09-20 14:46:25 +08:00
parent bbdcfbac0a
commit e009ea3036
3 changed files with 285 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/http_upstream.go
+9 -2
View File
@@ -76,6 +76,9 @@ const (
defaultOpenAIHTTP2FallbackErrorThreshold = 2
defaultOpenAIHTTP2FallbackWindow = 60 * time.Second
defaultOpenAIHTTP2FallbackTTL = 10 * time.Minute
// OpenAI 保留原有的探测与应答期限，避免中转站的延迟 PING 应答被长流策略提前判死。
openAIHTTP2ReadIdleTimeout = 15 * time.Second
openAIHTTP2PingTimeout = 15 * time.Second
// 长流 HTTP/2 连接健康探测：池化连接被代理/NAT
// 静默掐断会成为“死连接”（两端都以为存活），请求落上去会挂到 TCP 重传超时
// （分钟级）。Go 的 http2.Transport 默认 ReadIdleTimeout=0（不发健康 PING），
@@ -1340,7 +1343,7 @@ func buildUpstreamTransport(settings poolSettings, proxyURL *url.URL, protocolMo
transport.ForceAttemptHTTP2 = true
// 显式配置 http2 并启用 PING 健康探测，剔除代理/NAT 静默掐断的死连接，
// 避免请求挂在死连接上直到 TCP 重传超时（分钟级）。
if _, err := enableHTTP2KeepAlive(transport); err != nil {
if _, err := enableHTTP2KeepAlive(transport, protocolMode); err != nil {
return nil, err
}
case upstreamProtocolModeOpenAIH1:
@@ -1361,7 +1364,7 @@ func buildUpstreamTransport(settings poolSettings, proxyURL *url.URL, protocolMo
// Go 默认惰性配置 http2 且 ReadIdleTimeout=0（不发健康 PING），无法检测被代理/NAT
// 静默掐断的死连接。此处主动设置 ReadIdleTimeout/PingTimeout，让死连接被提前 PING
// 出并关闭，请求得以重建连接而非挂到 TCP 重传超时。返回底层 *http2.Transport 便于测试。
func enableHTTP2KeepAlive(transport *http.Transport) (*http2.Transport, error) {
func enableHTTP2KeepAlive(transport *http.Transport, protocolMode string) (*http2.Transport, error) {
h2, err := http2.ConfigureTransports(transport)
if err != nil {
return nil, err
@@ -1369,6 +1372,10 @@ func enableHTTP2KeepAlive(transport *http.Transport) (*http2.Transport, error) {
if h2 != nil {
h2.ReadIdleTimeout = longStreamHTTP2ReadIdleTimeout
h2.PingTimeout = longStreamHTTP2PingTimeout
if protocolMode == upstreamProtocolModeOpenAIH2 {
h2.ReadIdleTimeout = openAIHTTP2ReadIdleTimeout
h2.PingTimeout = openAIHTTP2PingTimeout
}
}
return h2, nil
}
backend/internal/repository/http_upstream_http2_keepalive_test.go
+71 -43
View File
@@ -2,6 +2,7 @@ package repository
import (
"context"
"crypto/tls"
"crypto/x509"
"net/http"
"net/http/httptest"
@@ -38,16 +39,24 @@ func requireHTTP2Configured(t *testing.T, tr *http.Transport, msg string) {
// 必须显式启用主动 PING 探测，让死连接被提前剔除，而不是只靠 ResponseHeaderTimeout
// 事后兜底。
func TestEnableHTTP2KeepAlive_EnablesPingHealthCheck(t *testing.T) {
tr := &http.Transport{}
h2, err := enableHTTP2KeepAlive(tr)
require.NoError(t, err)
require.NotNil(t, h2, "必须返回已配置的 *http2.Transport")
require.Positive(t, h2.ReadIdleTimeout, "必须启用空闲 PING 探测以剔除死连接")
require.Equal(t, longStreamHTTP2ReadIdleTimeout, h2.ReadIdleTimeout)
require.Equal(t, longStreamHTTP2PingTimeout, h2.PingTimeout, "PING 无响应必须有超时判定")
requireHTTP2Configured(t, tr, "http2 必须已挂到底层 http.Transport 上")
for _, tc := range []struct {
mode string
readIdleTimeout time.Duration
pingTimeout time.Duration
}{
{upstreamProtocolModeOpenAIH2, 15 * time.Second, 15 * time.Second},
{upstreamProtocolModeLongStreamH2, 10 * time.Second, 5 * time.Second},
} {
t.Run(tc.mode, func(t *testing.T) {
tr := &http.Transport{}
h2, err := enableHTTP2KeepAlive(tr, tc.mode)
require.NoError(t, err)
require.NotNil(t, h2, "必须返回已配置的 *http2.Transport")
require.Equal(t, tc.readIdleTimeout, h2.ReadIdleTimeout)
require.Equal(t, tc.pingTimeout, h2.PingTimeout, "各模式应使用独立的 PING 应答期限")
requireHTTP2Configured(t, tr, "http2 必须已挂到底层 http.Transport 上")
})
}
}
// long_stream_h2 模式构建的 Transport 必须带上 H2 PING 健康探测，从源头剔除死连接。
@@ -58,51 +67,70 @@ func TestBuildUpstreamTransport_LongStreamH2_EnablesPingHealthCheck(t *testing.T
requireHTTP2Configured(t, tr, "long_stream_h2 必须显式配置 http2 以启用 ReadIdleTimeout")
}
// 非 H2 模式（default/h1）不应因本次改动被误配置：default 走 Go 自动 H2（惰性配置，
// 构建时 Protocols/TLSNextProto 仍为空），h1 模式显式禁用 H2。避免波及 Claude/Gemini 热路径。
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")
// 默认、Grok 和显式 H1 模式不应主动启用 HTTP/2 保活，避免影响其他平台的传输策略。
func TestBuildUpstreamTransport_NonHTTP2_NotEagerlyConfigured(t *testing.T) {
for _, mode := range []string{upstreamProtocolModeDefault, upstreamProtocolModeGrok, upstreamProtocolModeOpenAIH1, upstreamProtocolModeOpenAIH1Fallback} {
t.Run(mode, func(t *testing.T) {
tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, mode)
require.NoError(t, err)
require.Nil(t, tr.Protocols, "非 H2 模式不应主动配置 http2 keepalive")
require.Nil(t, tr.TLSNextProto["h2"], "非 H2 模式不应主动配置 http2 keepalive")
})
}
}
// long_stream_h2 模式构建的 Transport 必须真正以 HTTP/2 与上游通信，PING 健康探测才有载体：
// 自定义 DialContext 下 Go 不会自动启用 H2，全靠 enableHTTP2KeepAlive 的显式配置。
func TestBuildUpstreamTransport_LongStreamH2_NegotiatesHTTP2(t *testing.T) {
// 使用同时支持 H1/H2 的 TLS 上游核对实际协议，确保保活配置不会改变其他模式的协商行为。
func TestBuildUpstreamTransport_NegotiatesExpectedProtocol(t *testing.T) {
srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
}))
srv.EnableHTTP2 = true
srv.StartTLS()
defer srv.Close()
tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, upstreamProtocolModeLongStreamH2)
require.NoError(t, err)
defer tr.CloseIdleConnections()
require.NotNil(t, tr.TLSClientConfig)
roots := x509.NewCertPool()
roots.AddCert(srv.Certificate())
tr.TLSClientConfig.RootCAs = roots
req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL, nil)
require.NoError(t, err)
resp, err := tr.RoundTrip(req)
require.NoError(t, err)
require.NoError(t, resp.Body.Close())
require.Equal(t, http.StatusOK, resp.StatusCode)
require.Equal(t, 2, resp.ProtoMajor, "long_stream_h2 必须协商到 HTTP/2")
for _, tc := range []struct {
mode string
wantMajor int
}{
{upstreamProtocolModeOpenAIH2, 2},
{upstreamProtocolModeLongStreamH2, 2},
{upstreamProtocolModeDefault, 1},
{upstreamProtocolModeGrok, 1},
{upstreamProtocolModeOpenAIH1, 1},
{upstreamProtocolModeOpenAIH1Fallback, 1},
} {
t.Run(tc.mode, func(t *testing.T) {
tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, tc.mode)
require.NoError(t, err)
defer tr.CloseIdleConnections()
roots := x509.NewCertPool()
roots.AddCert(srv.Certificate())
if tr.TLSClientConfig == nil {
tr.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12}
}
tr.TLSClientConfig.RootCAs = roots
req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL, nil)
require.NoError(t, err)
resp, err := tr.RoundTrip(req)
require.NoError(t, err)
require.NoError(t, resp.Body.Close())
require.Equal(t, http.StatusOK, resp.StatusCode)
require.Equal(t, tc.wantMajor, resp.ProtoMajor)
})
}
}
// 死连接在经 HTTP 代理（CONNECT 隧道）时最高发，这是带 proxy 账号的真实生产路径：
// 显式 http2 配置须与 Transport.Proxy 同时正确生效，不能相互干扰。
func TestBuildUpstreamTransport_LongStreamH2_WithHTTPProxy_EnablesKeepAlive(t *testing.T) {
func TestBuildUpstreamTransport_HTTP2_WithHTTPProxy_EnablesKeepAlive(t *testing.T) {
proxyURL, err := url.Parse("http://127.0.0.1:8080")
require.NoError(t, err)
tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), proxyURL, upstreamProtocolModeLongStreamH2)
require.NoError(t, err)
require.True(t, tr.ForceAttemptHTTP2)
requireHTTP2Configured(t, tr, "经代理的 long_stream_h2 也必须启用 http2 keepalive")
require.NotNil(t, tr.Proxy, "HTTP 代理仍须通过 Transport.Proxy 生效")
for _, mode := range []string{upstreamProtocolModeOpenAIH2, upstreamProtocolModeLongStreamH2} {
t.Run(mode, func(t *testing.T) {
tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), proxyURL, mode)
require.NoError(t, err)
require.True(t, tr.ForceAttemptHTTP2)
requireHTTP2Configured(t, tr, "经代理的 H2 模式也必须启用 http2 keepalive")
require.NotNil(t, tr.Proxy, "HTTP 代理仍须通过 Transport.Proxy 生效")
})
}
}
backend/internal/repository/http_upstream_http2_ping_test.go
+205
View File
@@ -0,0 +1,205 @@
package repository
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"strings"
"testing"
"testing/synctest"
"time"
"github.com/stretchr/testify/require"
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
)
// 使用真实传输构建器和 HTTP/2 帧验证保活行为；虚拟时间保持生产超时值且无需实际等待。
func TestBuildUpstreamTransport_HTTP2PingPolicy(t *testing.T) {
for _, tc := range []struct {
name string
mode string
ackDelay time.Duration
finishDelay time.Duration
headersEarly bool
wantLost bool
wantElapsed time.Duration
}{
{"OpenAI延迟应答", upstreamProtocolModeOpenAIH2, 6 * time.Second, 22 * time.Second, false, false, 22 * time.Second},
{"OpenAI已开始输出后延迟应答", upstreamProtocolModeOpenAIH2, 6 * time.Second, 22 * time.Second, true, false, 22 * time.Second},
{"OpenAI正常应答但长时间无业务输出", upstreamProtocolModeOpenAIH2, 0, 40 * time.Second, false, false, 40 * time.Second},
{"OpenAI无应答且未收到响应头", upstreamProtocolModeOpenAIH2, -1, time.Minute, false, true, 30 * time.Second},
{"OpenAI无应答且已开始输出", upstreamProtocolModeOpenAIH2, -1, time.Minute, true, true, 30 * time.Second},
{"长流保持原有超时", upstreamProtocolModeLongStreamH2, 6 * time.Second, 22 * time.Second, false, true, 15 * time.Second},
{"长流正常应答", upstreamProtocolModeLongStreamH2, 0, 40 * time.Second, false, false, 40 * time.Second},
} {
t.Run(tc.name, func(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
clientConn, serverConn := net.Pipe()
defer func() { require.NoError(t, clientConn.Close()) }()
defer func() { require.NoError(t, serverConn.Close()) }()
peerDone := make(chan error, 1)
go func() {
err := serveHTTP2PingTestPeer(serverConn, tc.ackDelay, tc.finishDelay, tc.headersEarly)
// 业务结果读完后客户端会关闭连接，此时尚在发送的 PING 应答可正常退出。
if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || errors.Is(err, io.ErrClosedPipe) {
err = nil
}
peerDone <- err
}()
tr, err := buildUpstreamTransport(http2KeepAliveTestPoolSettings(), nil, tc.mode)
require.NoError(t, err)
defer tr.CloseIdleConnections()
// 仅把 TLS 建连替换为内存 HTTP/2 连接，保活参数仍来自生产构建器。
tr.Protocols.SetHTTP1(false)
tr.Protocols.SetUnencryptedHTTP2(true)
tr.DialContext = func(context.Context, string, string) (net.Conn, error) {
return clientConn, nil
}
client := &http.Client{Transport: tr, Timeout: 50 * time.Second}
started := time.Now()
resp, requestErr := client.Post("http://upstream.test/v1/responses", "application/json", strings.NewReader(`{"stream":true}`))
resultErr := requestErr
var body []byte
if resp != nil {
require.Equal(t, http.StatusOK, resp.StatusCode)
require.Equal(t, 2, resp.ProtoMajor)
body, resultErr = io.ReadAll(resp.Body)
require.NoError(t, resp.Body.Close())
}
if tc.wantLost {
require.ErrorContains(t, resultErr, "http2: client connection lost")
if tc.headersEarly {
require.NoError(t, requestErr)
require.Equal(t, "data: first\n\n", string(body))
} else {
require.Nil(t, resp)
}
} else {
require.NoError(t, resultErr)
require.NotNil(t, resp)
wantBody := "data: [DONE]\n\n"
if tc.headersEarly {
wantBody = "data: first\n\n" + wantBody
}
require.Equal(t, wantBody, string(body))
}
require.Equal(t, tc.wantElapsed, time.Since(started))
require.NoError(t, clientConn.Close())
require.NoError(t, <-peerDone)
})
})
}
}
// 对端持续读取请求，只对 PING 的应答施加延迟；业务响应独立于 PING 定时发送。
func serveHTTP2PingTestPeer(conn net.Conn, ackDelay, finishDelay time.Duration, headersEarly bool) error {
framer := http2.NewFramer(conn, conn)
type readResult struct {
frame http2.Frame
err error
}
// 缓冲握手帧，避免 net.Pipe 双向无缓冲写入造成 SETTINGS 握手互等。
frames := make(chan readResult, 16)
done := make(chan struct{})
defer close(done)
go func() {
preface := make([]byte, len(http2.ClientPreface))
_, err := io.ReadFull(conn, preface)
if err == nil && string(preface) != http2.ClientPreface {
err = errors.New("HTTP/2 客户端前言不匹配")
}
for {
var frame http2.Frame
if err == nil {
frame, err = framer.ReadFrame()
}
select {
case frames <- readResult{frame, err}:
case <-done:
return
}
if err != nil {
return
}
}
}()
if err := framer.WriteSettings(); err != nil {
return err
}
sendHeaders := func(streamID uint32) error {
var block bytes.Buffer
encoder := hpack.NewEncoder(&block)
for _, field := range []hpack.HeaderField{{Name: ":status", Value: "200"}, {Name: "content-type", Value: "text/event-stream"}} {
if err := encoder.WriteField(field); err != nil {
return err
}
}
return framer.WriteHeaders(http2.HeadersFrameParam{StreamID: streamID, BlockFragment: block.Bytes(), EndHeaders: true})
}
var finishCh, ackCh <-chan time.Time
var streamID uint32
var pingData [8]byte
for {
select {
case result := <-frames:
if errors.Is(result.err, io.EOF) || errors.Is(result.err, net.ErrClosed) {
return nil
}
if result.err != nil {
return result.err
}
switch frame := result.frame.(type) {
case *http2.SettingsFrame:
if !frame.IsAck() {
if err := framer.WriteSettingsAck(); err != nil {
return err
}
}
case *http2.HeadersFrame:
if streamID != 0 {
return fmt.Errorf("意外重放请求：流 %d", frame.StreamID)
}
streamID = frame.StreamID
if headersEarly {
if err := sendHeaders(streamID); err != nil {
return err
}
if err := framer.WriteData(streamID, false, []byte("data: first\n\n")); err != nil {
return err
}
}
finishTimer := time.NewTimer(finishDelay)
defer finishTimer.Stop()
finishCh = finishTimer.C
case *http2.PingFrame:
if !frame.IsAck() && ackDelay >= 0 {
pingData = frame.Data
ackTimer := time.NewTimer(ackDelay)
defer ackTimer.Stop()
ackCh = ackTimer.C
}
}
case <-ackCh:
ackCh = nil
if err := framer.WritePing(true, pingData); err != nil {
return err
}
case <-finishCh:
finishCh = nil
if !headersEarly {
if err := sendHeaders(streamID); err != nil {
return err
}
}
if err := framer.WriteData(streamID, true, []byte("data: [DONE]\n\n")); err != nil {
return err
}
}
}
}