diff --git a/backend/internal/repository/http_upstream.go b/backend/internal/repository/http_upstream.go index e073659cc2..f765d0f66e 100644 --- a/backend/internal/repository/http_upstream.go +++ b/backend/internal/repository/http_upstream.go @@ -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 } diff --git a/backend/internal/repository/http_upstream_http2_keepalive_test.go b/backend/internal/repository/http_upstream_http2_keepalive_test.go index 29acf4b631..249f246ff8 100644 --- a/backend/internal/repository/http_upstream_http2_keepalive_test.go +++ b/backend/internal/repository/http_upstream_http2_keepalive_test.go @@ -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 生效") + }) + } } diff --git a/backend/internal/repository/http_upstream_http2_ping_test.go b/backend/internal/repository/http_upstream_http2_ping_test.go new file mode 100644 index 0000000000..2f7f5cc71e --- /dev/null +++ b/backend/internal/repository/http_upstream_http2_ping_test.go @@ -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 + } + } + } +}