github/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-07 13:48:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge PR #6306

Browse Source 
fix(openai): emit pre-output SSE keepalive on the /v1/responses passthrough path
This commit is contained in:
Wesley Liddick
2026-08-29 09:04:46 +08:00
committed by GitHub
parent 3045b3ade1 60756c0ca4
commit 46598dd491
3 changed files with 164 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/service/openai_compact_sse_keepalive.go
+14 -1
View File
@@ -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
backend/internal/service/openai_gateway_passthrough.go
+43
View File
@@ -1808,6 +1808,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 := ""
@@ -2060,6 +2098,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
backend/internal/service/openai_passthrough_preoutput_keepalive_test.go
+107
View File
@@ -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
}