From 60756c0ca438693f6badcf5c20a5120b98a4909b Mon Sep 17 00:00:00 2001 From: HuntercodeT <61233559+HuntercodeT@users.noreply.github.com> Date: Fri, 28 Aug 2026 01:47:33 +0000 Subject: [PATCH] fix(openai): emit pre-output SSE keepalive on the /v1/responses passthrough path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The passthrough streaming loop buffers leading events in `pendingLines` until the first visible output, so a client receives **zero bytes** — not even the HTTP response headers, since gin's ResponseWriter only commits them on the first write — for as long as the upstream takes to produce its first visible event. The Forward path already recognises this exact hazard and solves it with a downstream keepalive: // Track downstream writes separately from upstream reads: pre-output // failover can buffer response.created / response.in_progress, so // keepalive must be based on downstream idle time. The passthrough path has the same buffering but never starts a keepalive. Reasoning models routinely think for several minutes before emitting a visible event, so intermediate proxies close the connection on idle timeout. Production evidence from one deployment (12h window, gpt-5.6 family): * 44 `/v1/responses` requests produced their first visible output only after 600–900 s, each carrying 30k–50k output tokens — the upstream had finished the work. * Every one of them was killed by an intermediate nginx `proxy_read_timeout` (600 s) and returned 504. The user received nothing. * Cutting early does not even save resources: the code deliberately keeps draining the upstream for usage after the client disconnects. Fix: start the existing keepalive helper for the passthrough stream as well. * `StartOpenAICompactSSEKeepalive` keeps its compact-marker gate; the gate-free body is extracted as `startOpenAISSEKeepalive` for callers that already know they are in an SSE context. * The passthrough loop starts it when `gateway.stream_keepalive_interval > 0` and stops it immediately before its first real write, so the ResponseWriter is never shared. Pre-output failover is unaffected: keepalive bytes are already excluded by `OpenAICompactKeepaliveAdjustedWrittenSize`, which `openAIStreamClientOutputStarted` on this very path already consults (#3887). No pendingLines and no account-specific headers are ever emitted by a keepalive. Behaviour is unchanged when `gateway.stream_keepalive_interval` is 0. --- .../service/openai_compact_sse_keepalive.go | 15 ++- .../service/openai_gateway_passthrough.go | 43 +++++++ ...ai_passthrough_preoutput_keepalive_test.go | 107 ++++++++++++++++++ 3 files changed, 164 insertions(+), 1 deletion(-) create mode 100644 backend/internal/service/openai_passthrough_preoutput_keepalive_test.go diff --git a/backend/internal/service/openai_compact_sse_keepalive.go b/backend/internal/service/openai_compact_sse_keepalive.go index 360c1d55a5..c0aba9108b 100644 --- a/backend/internal/service/openai_compact_sse_keepalive.go +++ b/backend/internal/service/openai_compact_sse_keepalive.go @@ -43,7 +43,20 @@ type openAICompactSSEKeepalive struct { // 响应构造都会先在心跳互斥锁下停拍，未被显式拦截的写回路径（如 Forward // 内部的本地拒绝）也不会与心跳 goroutine 产生数据竞争或字节交错。 func StartOpenAICompactSSEKeepalive(c *gin.Context, interval time.Duration) func() { - if c == nil || c.Writer == nil || interval <= 0 || !openAICompactClientWantsStream(c) { + if !openAICompactClientWantsStream(c) { + return func() {} + } + return startOpenAISSEKeepalive(c, interval) +} + +// startOpenAISSEKeepalive 是不检查 compact 标记的内部入口，供【已经确定处于 +// SSE 流式上下文】的调用方使用（例如 /v1/responses 透传：进入流式循环时上游 +// 已返回 text/event-stream，SSE 响应头也已设好）。 +// +// 心跳字节由 OpenAICompactKeepaliveAdjustedWrittenSize 统一排除，因此不会污染 +// "是否已向客户端写出语义响应"的 failover 判定（见 #3887）。 +func startOpenAISSEKeepalive(c *gin.Context, interval time.Duration) func() { + if c == nil || c.Writer == nil || interval <= 0 { return func() {} } originalWriter := c.Writer diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index e3f19f335e..b418288347 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -1749,6 +1749,44 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( upstreamRequestID := strings.TrimSpace(resp.Header.Get("x-request-id")) // pendingLines 在首个可见输出前保留前导事件，确保无输出失败仍可安全 failover。 pendingLines := make([]string, 0, 8) + + // ── 首个可见输出之前的下游 keepalive ────────────────────────────────── + // + // 与 Forward 路径同源的问题。openai_gateway_response_handling.go 里那句注释 + // 说得最清楚： + // + // "Track downstream writes separately from upstream reads: pre-output + // failover can buffer response.created / response.in_progress, so + // keepalive must be based on downstream idle time." + // + // 上面的 pendingLines 正是同一种缓冲：首个可见输出到来之前，下游【一个字节 + // 都收不到】—— 连 HTTP 响应头都不会提交（gin 的 ResponseWriter 直到首次写入 + // 才发送 header）。Forward 路径为此加了心跳，透传路径漏了。 + // + // 推理模型在首个可见输出前思考数百秒是常态，于是中间层代理会按空闲超时把 + // 连接判死。这不是假设：某生产部署实测 12 小时内 44 个 /v1/responses 请求在 + // 600~900s 才产出首个可见输出（每个 3~5 万 output token，上游其实算完了）， + // 全部被中间 nginx 的 proxy_read_timeout(600s) 判超时回 504，用户一个字没拿到。 + // + // 心跳写出的 SSE 注释同时做到三件事： + // 1. 提交 HTTP 响应头，让下游知道连接活着； + // 2. 刷新中间层的空闲超时（proxy_read_timeout 衡量的是两次读之间的间隔， + // 不是请求总时长），长推理因此不再被误杀； + // 3. 不写出任何 pendingLines、不泄露账号相关的头， + // 且心跳字节已由 OpenAICompactKeepaliveAdjustedWrittenSize 排除， + // 所以 pre-output failover 的能力完全不受影响（#3887 的记账在此复用）。 + // + // 用 startOpenAISSEKeepalive 而不是 StartOpenAICompactSSEKeepalive：后者会检查 + // compact 标记，而这里是普通 /v1/responses 透传。走到这一行时上游已回 + // text/event-stream、SSE 响应头也已设好，处于流式上下文是确定的。 + stopKeepalive := func() {} + if s.cfg != nil && s.cfg.Gateway.StreamKeepaliveInterval > 0 { + stopKeepalive = startOpenAISSEKeepalive(c, + time.Duration(s.cfg.Gateway.StreamKeepaliveInterval)*time.Second) + } + // 任何返回路径都要停拍。Stop 与心跳 goroutine 之间有互斥锁， + // 返回后不会再有字节写出。 + defer stopKeepalive() // flushPending 表示已写入但未到 SSE 空行边界的脏状态；defer 兜底函数退出前的残留，断连后不再 Flush。 flushPending := false pendingSSEEventType := "" @@ -2001,6 +2039,11 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( pendingLines = append(pendingLines, line) continue } + // 真实输出开始，心跳的使命结束。停拍是幂等的，且会与心跳 goroutine + // 建立 happens-before —— 之后 ResponseWriter 由本循环独占。 + if !clientOutputStarted { + stopKeepalive() + } if !clientOutputStarted && len(pendingLines) > 0 { if !writePendingLines() { continue diff --git a/backend/internal/service/openai_passthrough_preoutput_keepalive_test.go b/backend/internal/service/openai_passthrough_preoutput_keepalive_test.go new file mode 100644 index 0000000000..b1ee6b9b6a --- /dev/null +++ b/backend/internal/service/openai_passthrough_preoutput_keepalive_test.go @@ -0,0 +1,107 @@ +package service + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +// 这一组守住的是「首个可见输出之前下游不再静默」这条性质。 +// +// 透传路径的 pendingLines 会把 response.created / response.in_progress 全部扣住， +// 于是首个可见输出之前下游一个字节都收不到，连 HTTP 响应头都不会提交。推理模型 +// 思考数百秒时，中间层代理会按空闲超时把连接判死。Forward 路径早就用心跳解决了 +// 这个问题（见 openai_gateway_response_handling.go 中 lastDownstreamWriteAt 的注释）， +// 透传路径漏了。 + +func newPassthroughKeepaliveTestContext(t *testing.T) (*gin.Context, *httptest.ResponseRecorder) { + t.Helper() + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + // 刻意【不】调用 MarkOpenAICompactClientStream：普通 /v1/responses 透传不带 + // compact 标记，这正是它此前拿不到心跳的原因。 + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + return c, rec +} + +// startOpenAISSEKeepalive 必须在【没有】compact 标记时也能启动 —— +// 否则普通透传请求依旧静默。 +func TestStartOpenAISSEKeepalive_WorksWithoutCompactMarker(t *testing.T) { + c, rec := newPassthroughKeepaliveTestContext(t) + + // 对照:带 compact 标记检查的入口在这里应当直接 no-op。 + stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval) + waitForKeepaliveBeats() + stop() + require.Zero(t, rec.Body.Len(), "无 compact 标记时 StartOpenAICompactSSEKeepalive 应当 no-op") + + // 内部入口不检查标记,应当真的开始打拍。 + c, rec = newPassthroughKeepaliveTestContext(t) + stop = startOpenAISSEKeepalive(c, keepaliveTestInterval) + defer stop() + waitForKeepaliveBeats() + + require.True(t, StopOpenAICompactSSEKeepaliveCommitted(c), "心跳应当提交响应头") + require.Equal(t, http.StatusOK, rec.Code) + require.Equal(t, "text/event-stream", rec.Header().Get("Content-Type")) + require.Equal(t, "no", rec.Header().Get("X-Accel-Buffering")) + require.Contains(t, rec.Body.String(), ": keepalive\n\n") +} + +// 🔴 最要紧的一条:心跳字节【不得】把请求判成「已向客户端写出语义响应」， +// 否则上游 429/5xx 时不再换号 —— 这正是 #3887 加固的那条不变量， +// 透传路径的 pre-output failover 完全依赖它。 +func TestPassthroughKeepaliveDoesNotBlockPreOutputFailover(t *testing.T) { + c, rec := newPassthroughKeepaliveTestContext(t) + stop := startOpenAISSEKeepalive(c, keepaliveTestInterval) + defer stop() + waitForKeepaliveBeats() + require.True(t, StopOpenAICompactSSEKeepaliveCommitted(c)) + require.NotZero(t, rec.Body.Len(), "前提:心跳确实写出了字节") + + // 只有心跳字节时,仍应判定为「尚未向客户端输出」。 + require.False(t, openAIStreamClientOutputStarted(c, false), + "心跳字节不构成语义输出,pre-output failover 必须仍然可用") + + // 写出一条真实事件之后,判定才翻转。 + _, err := c.Writer.Write([]byte("data: {\"type\":\"response.output_text.delta\"}\n\n")) + require.NoError(t, err) + require.True(t, openAIStreamClientOutputStarted(c, false), + "真实语义输出之后应当判定为已输出") +} + +// 停拍之后不得再有心跳字节写出 —— 主循环接管 ResponseWriter 的前提。 +func TestPassthroughKeepaliveStopsBeforeHandingOverWriter(t *testing.T) { + c, rec := newPassthroughKeepaliveTestContext(t) + stop := startOpenAISSEKeepalive(c, keepaliveTestInterval) + waitForKeepaliveBeats() + stop() + + before := rec.Body.String() + waitForKeepaliveBeats() + require.Equal(t, before, rec.Body.String(), "停拍后不应再有字节写出") + + // 停拍后主循环写出的内容不应被心跳穿插。 + _, err := c.Writer.Write([]byte("data: real\n\n")) + require.NoError(t, err) + waitForKeepaliveBeats() + require.True(t, strings.HasSuffix(rec.Body.String(), "data: real\n\n"), + "停拍后写入应当是响应体的最后一段") +} + +// interval<=0(配置禁用)时行为与改动前完全一致:一个字节都不写。 +func TestPassthroughKeepaliveDisabledKeepsWriterUntouched(t *testing.T) { + c, rec := newPassthroughKeepaliveTestContext(t) + stop := startOpenAISSEKeepalive(c, 0) + waitForKeepaliveBeats() + stop() + require.Zero(t, rec.Body.Len()) + require.False(t, StopOpenAICompactSSEKeepaliveCommitted(c)) + _ = time.Now +}