From 7457b052a725b1b8d634ef36a300b3910270709b Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sat, 24 Oct 2026 12:35:31 +0800 Subject: [PATCH] =?UTF-8?q?fix(webui):=20streamOpenAI=20=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=20w=20=E5=8D=95=E5=86=99=E8=80=85=E6=A8=A1=E5=9E=8B=EF=BC=8C?= =?UTF-8?q?=E4=BF=AE=20gzipResponseWriter=20=E7=9C=9F=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E7=AB=9E=E4=BA=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因（-race 实测，非推测）： streamOpenAI 的 handler goroutine 在收尾处直接 fmt.Fprintf(w, finish chunk)、 fmt.Fprintf(w, "data: [DONE]")、flusher.Flush()，与增量 flush goroutine （16ms 批量合并）**并发写同一个 http.ResponseWriter**。经 gzip 中间件时， gzipResponseWriter 的 mode / wroteHeader / buf 三个字段无锁被两方同时读写： Read gzipResponseWriter.Write gzip.go:139（读 g.mode） └─ streamOpenAI handler goroutine handler_openai.go:256 Write gzipResponseWriter.commit gzip.go:206（写 g.mode/wroteHeader） └─ streamOpenAI 写 goroutine handler_openai.go:165 修法选择：改写并发模型（单写者），不给 gzipResponseWriter 加锁。 理由两条： 1. 加锁只哄住 race detector，是治标 —— 并发写同一个 ResponseWriter 这件事 本身就该被结构禁止，而不是靠锁逐字段兜。 2. 持锁要跨 Write / Flush 这类网络 IO，一旦客户端慢，整条 SSE 会串行卡在 锁上（锁内做 IO）。单写者模型从结构上消灭竞争，且零锁开销。 改动： - 新增 finishCh（chan finishItem，buf=1）+ writerExit。终止帧（finish chunk 或超时错误 + [DONE] + Flush）也交给那个唯一的写 goroutine 发。 - 新增 sendFinish(usage, timedOut)：**先投递再等 writerExit**。 顺序不能反 —— writer goroutine 只在收到 finishCh 或 done 关闭时退出，而 done 在 defer 里关（函数返回之后），先等后发即死锁。select 里的 <-done 是防御路径：done 已关时 writer 可能已不在，不该再等一个永不到来的接收者。 - 超时路径与正常收尾路径都改走 sendFinish，handler goroutine 不再碰 w。 - writerExit 用 safeclose.Safe 关闭，遵守本仓 close 纪律 （internal/safeclose/TestNoBareCloseOutsideOutsideSafeclose 禁裸 close(chan)）。 - 顺带修掉一个既有逻辑缺陷：旧写法下 [DONE] 可能先于 deltaCh 里尚未发完的 尾部 delta 到达客户端；现在先 flushPending 再发终止帧。 验证（全部 -count=1，除注明者）： - 基线（f408448 原版）TestOpenAIStreamIsActuallyStreaming + -race： 单跑 5/5 次各 8 条 DATA RACE 且 FAIL；-count=5 时 10 / 12 / 10 条。 - 修复后同一用例 + -race：单跑 6/6 次 races=0 PASS。 - 修复后整包 go test -race ./internal/plugins/webui/：3/3 次 races=0 PASS （63s / 69s / 64s）。 - go build ./... 通过；go vet ./... 通过；go test ./... 58 包 ok、0 FAIL、 0 WARNING: DATA RACE。 判据：cr-HomeAgent-6c0d1e07（stress），带变异 M1「sendFinish 改回直写 w （并吃掉 <-writerExit，避免只靠死锁超时报红）」——实测被咬住（判据报红）。 --- internal/plugins/webui/handler_openai.go | 67 +++++++++++++++++++++--- 1 file changed, 59 insertions(+), 8 deletions(-) diff --git a/internal/plugins/webui/handler_openai.go b/internal/plugins/webui/handler_openai.go index b79d7c60..c92963c1 100644 --- a/internal/plugins/webui/handler_openai.go +++ b/internal/plugins/webui/handler_openai.go @@ -152,10 +152,28 @@ func (h *Handler) streamOpenAI(ctx context.Context, w http.ResponseWriter, model // 写 goroutine：批量合并（16ms 窗口）后 flush，降 syscall 次数。 // 不 close(deltaCh) —— 订阅回调可能在 handler 返回后仍被总线异步触发， // close 后再发会 panic（生产日志里出现过单日数千次）。 + // + // ★★ 并发契约：本函数内 w 只有一个写者 = 下面这个 writer goroutine。 + // 此前 handler goroutine 在收尾处直写 w（finish chunk/[DONE]/Flush）， + // 与这里的写 goroutine 并发写同一个 http.ResponseWriter —— gzip 中间件 + // 的 mode/wroteHeader/buf 被两个 goroutine 同时读写，-race 单跑即可 + // 稳定复现 6~8 条 DATA RACE。修法选「单写者模型」而非给 + // gzipResponseWriter 加锁：加锁只哄住 race detector（治标），且持锁跨 + // 网络 Write/Flush 会把整条 SSE 卡在一个串行点上；「同一时刻只有一个 + // goroutine 写 w」才是 SSE 帧完整性的正确约束。于是终止帧也必须交给 + // 这个唯一的写者：见 finishCh / sendFinish。 + finishCh := make(chan finishItem, 1) + writerExit := make(chan struct{}) + // 用 safeclose 而不是裸 close：本仓 close 纪律测试 + // （internal/safeclose/TestNoBareCloseOutsideSafeclose）禁止裸 close(chan)， + // 理由是「今天只关一次的 channel，明天多一条退出路径就变成关两次 ⇒ panic」。 + var writerExitOnce sync.Once var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() + // 终止帧全部写完（含 Flush）后才关闭，handler 侧据此确认已出网。 + defer safeclose.Safe(&writerExitOnce, writerExit) pending := make([]string, 0, 64) flushPending := func() { if len(pending) == 0 { @@ -188,6 +206,19 @@ func (h *Handler) streamOpenAI(ctx context.Context, w http.ResponseWriter, model } case <-flushTicker.C: flushPending() + case f := <-finishCh: + // 先把攒着的尾部 delta 发完，再发终止帧 —— 顺序不能反， + // 否则客户端会先看到 [DONE] 再看到内容（旧写法的缺陷）。 + flushPending() + if f.timedOut { + fmt.Fprintf(w, "data: %s\n\n", `{"error":{"message":"agent timeout (300s)"}}`) + } else { + fmt.Fprintf(w, "%s\n", openAIChunkLine(completionID, created, model, + map[string]interface{}{}, "stop", f.usage)) + } + fmt.Fprintf(w, "data: [DONE]\n\n") + flusher.Flush() + return case <-done: flushPending() return @@ -195,6 +226,21 @@ func (h *Handler) streamOpenAI(ctx context.Context, w http.ResponseWriter, model } }() + // sendFinish 把「终止」交给唯一写者，并等它把终止帧真正写完（含 Flush）。 + // 此后 handler goroutine 不再碰 w 一个字节。 + // + // 必须**先发再等**：writer goroutine 只在「收到 finishCh」或「done 关闭」 + // 时退出，而 done 的关闭在 defer 里（本函数返回之后）——先等退出后发 + // 终止就是死锁。select 的 <-done 是防御：done 已关时 writer 可能已不在， + // 此时不能再等一个永不到来的接收者。 + sendFinish := func(usage map[string]interface{}, timedOut bool) { + select { + case finishCh <- finishItem{usage: usage, timedOut: timedOut}: + case <-done: + } + <-writerExit + } + send := func(d deltaItem) { select { case deltaCh <- d: @@ -241,9 +287,8 @@ func (h *Handler) streamOpenAI(ctx context.Context, w http.ResponseWriter, model select { case response = <-respCh: case <-ctx.Done(): - fmt.Fprintf(w, "data: %s\n\n", `{"error":{"message":"agent timeout (300s)"}}`) - fmt.Fprintf(w, "data: [DONE]\n\n") - flusher.Flush() + // 超时也走同一个唯一写者出口：在这里直写 w 正是竞争的另一半。 + sendFinish(nil, true) return } @@ -252,11 +297,8 @@ func (h *Handler) streamOpenAI(ctx context.Context, w http.ResponseWriter, model usage, _ = response.Payload["usage"].(map[string]interface{}) } - // finish chunk + [DONE] - fmt.Fprintf(w, "%s\n", openAIChunkLine(completionID, created, model, - map[string]interface{}{}, "stop", usage)) - fmt.Fprintf(w, "data: [DONE]\n\n") - flusher.Flush() + // finish chunk + [DONE]：交给唯一写者发。 + sendFinish(usage, false) } // openAIChunkLine 把一个 chunk 序列化成一行 SSE。 @@ -284,6 +326,15 @@ func openAIChunkLine(id string, created int64, model string, delta map[string]in return "data: " + string(data) } +// finishItem 是流终止消息：由 handler goroutine 投递、由唯一的 writer +// goroutine 消费并写出 finish chunk 与 [DONE]。 +type finishItem struct { + // usage 进 finish chunk（OpenAI 协议里 usage 只在收尾帧出现）。 + usage map[string]interface{} + // timedOut 为真表示走 ctx 超时收尾（在流里报错而不是发 stop）。 + timedOut bool +} + type deltaItem struct { content string reasoning string