Merge pull request #28 from JianFeeeee/fix/webui-gzip-race

fix(webui): streamOpenAI 改为 w 单写者模型，修 gzipResponseWriter 真数据竞争

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