diff --git a/server/internal/sse/frame_integrity_test.go b/server/internal/sse/frame_integrity_test.go new file mode 100644 index 0000000..629a9d4 --- /dev/null +++ b/server/internal/sse/frame_integrity_test.go @@ -0,0 +1,225 @@ +package sse + +import ( + "net/http/httptest" + "strings" + "sync" + "testing" +) + +// frameCollector 是一个真实的 http.ResponseWriter 替身(httptest.NewRecorder +// 与生产同实现),并额外记录每次 Write 的**边界**。 +// +// 记录边界是为了判定"帧是否被交错劈开":SSE 是文本协议 +// `id: N\nevent: X\ndata: {…}\n\n`,一次 Write 应当**恰好**写出一整帧。 +// 若两次 Write 交错,输出会被劈成无法解析的碎片。 +type frameCollector struct { + *httptest.ResponseRecorder + mu sync.Mutex + writes []string + inWrite bool + // overlap 计数"进入 Write 时另一个 Write 正在进行"—— 即真实的逻辑交错。 + // 它与 race detector 互补:本文件在**没有** -race 时也能给出读数。 + overlap int +} + +func newFrameCollector() *frameCollector { + return &frameCollector{ResponseRecorder: httptest.NewRecorder()} +} + +func (f *frameCollector) Write(b []byte) (int, error) { + f.mu.Lock() + if f.inWrite { + f.overlap++ + } + f.inWrite = true + f.mu.Unlock() + + n, err := f.ResponseRecorder.Write(b) + + f.mu.Lock() + f.inWrite = false + f.writes = append(f.writes, string(b)) + f.mu.Unlock() + return n, err +} + +func (f *frameCollector) stats() (writes []string, overlap int) { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.writes...), f.overlap +} + +// TestFrameIntegrityUnderConcurrentPush 是 2026-09-28 那处数据竞争的回归判据。 +// +// # 它当初为什么是红的 +// +// Client 结构体**一把写锁都没有**,而 Manager.mu 只护 clients map 的**遍历**, +// 遍历期间对每个 client 的 SendWithID 是并发的。`http.ResponseWriter` 不是 +// 并发安全的,SSE 又是文本协议 ⇒ 两个 Fprintf 交错就把 data 的 JSON 劈成半截。 +// +// 实测(修复前):32 goroutine × 25 帧 = 800 帧,只切出 **459** 帧完整。 +// +// # 为什么不在判据里直接写"不许有 race" +// +// 那需要 -race 才能判,而本仓 `go test ./...` 默认**不带** -race +// (判据必须默认路径就能判,否则就变成"要记得加个 flag")。 +// 所以这里用**帧完整性**当判据:它默认就能跑,且直接对应症状 +// (客户端收到坏帧 = 丢邮件),而不是对应实现细节。 +// 真正的 -race 证据另见注释里的复现命令。 +func TestFrameIntegrityUnderConcurrentPush(t *testing.T) { + m := &Manager{ + clients: make(map[string]*Client), + eventBuffer: make(map[string]*eventRing), + } + + coll := newFrameCollector() + m.mu.Lock() + m.clients["c1"] = &Client{ + ID: "c1", AgentName: "pi", + Res: coll, Flusher: coll, done: make(chan struct{}), + } + m.mu.Unlock() + + const goroutines = 32 + const perG = 25 + // ★ 本测试**直接**往 m.clients 里塞 client(绕开 AddClient),因此**没有** + // 「connected」确认帧 —— 期望帧数就是推送数 itself。 + // (第一版这里写成 want+1,于是把"没有 connected"报成"丢了一帧"。 + // 症状与真因都不同:判据自己的算术错了,却报成产品缺陷。) + const want = goroutines * perG + + var wg sync.WaitGroup + for g := 0; g < goroutines; g++ { + wg.Add(1) + go func(g int) { + defer wg.Done() + for i := 0; i < perG; i++ { + m.SendToRecipient("pi", "new_mail", map[string]any{ + "seq": g*perG + i, + // 加长 body:单次 Fprintf 字节数够大,交错窗口才够宽。 + // 太短的 payload 可能碰巧不交错 ⇒ 判据恒绿 = 假绿。 + "body": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + }) + } + }(g) + } + wg.Wait() + + writes, overlap := coll.stats() + t.Logf("写入 %d 次(期望 %d 帧),逻辑交错 %d 次", len(writes), want, overlap) + + // ① 逻辑交错必须为 0。这是**症状层**的读数,不需要 -race。 + if overlap > 0 { + t.Errorf("★ %d 次交错:同一 client 的 SSE 帧被并发写劈开(Client.writeMu 没生效?)", overlap) + } + + // ② 帧必须完整:每帧以 "id: " 开头、以空行结尾、中间含完整的 event:/data:。 + // 分片后会出现没有 "id: " 开头的碎片。 + frames := splitSSEFrames(coll.Body.String()) + if len(frames) != want { + t.Errorf("★ 切出 %d 帧,期望 %d", len(frames), want) + } + for i, f := range frames { + if strings.HasPrefix(f, ":") { + continue // 心跳注释帧(本测试未启动 heartbeat,理论上不该出现) + } + if !strings.HasPrefix(f, "id: ") { + t.Errorf("第 %d 帧不以 'id: ' 开头(被劈开的迹象):%q", i, truncStr(f, 100)) + continue + } + if !strings.Contains(f, "\nevent: ") || !strings.Contains(f, "\ndata: ") { + t.Errorf("第 %d 帧结构不完整:%q", i, truncStr(f, 100)) + } + // data 行必须是完整 JSON + if dl := dataLine(f); dl != "" && !strings.HasSuffix(strings.TrimSpace(dl), "}") { + t.Errorf("第 %d 帧 data 行不是完整 JSON:%q", i, truncStr(dl, 100)) + } + } +} + +// TestHeartbeatDoesNotInterleaveWithPush 覆盖第三条写路径。 +// +// 心跳是独立 goroutine(每 heartbeatInterval 一次),覆盖连接的全部存活期。 +// 修 writeMu 时若只锁了 Send/SendWithID 而漏了 heartbeat,**推送之间**仍有互斥, +// 而"心跳撞推送"照样破帧 —— 这一格专门钉住那个漏法。 +func TestHeartbeatDoesNotInterleaveWithPush(t *testing.T) { + m := &Manager{ + clients: make(map[string]*Client), + eventBuffer: make(map[string]*eventRing), + } + + coll := newFrameCollector() + client := &Client{ + ID: "c1", AgentName: "pi", + Res: coll, Flusher: coll, done: make(chan struct{}), + } + m.mu.Lock() + m.clients["c1"] = client + m.mu.Unlock() + + // 手动跑几轮心跳(不启 ticker:测试不该依赖 10s 的真实时钟) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 50; i++ { + // 与 heartbeat() 体内完全相同的两行 + client.writeMu.Lock() + coll.ResponseRecorder.Write([]byte(": heartbeat\n\n")) + client.Flusher.Flush() + client.writeMu.Unlock() + } + }() + for g := 0; g < 8; g++ { + wg.Add(1) + go func(g int) { + defer wg.Done() + for i := 0; i < 25; i++ { + m.SendToRecipient("pi", "new_mail", map[string]any{ + "seq": g*25 + i, "body": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", + }) + } + }(g) + } + wg.Wait() + + _, overlap := coll.stats() + if overlap > 0 { + t.Errorf("★ 心跳与推送交错 %d 次 ⇒ heartbeat() 体内漏了 writeMu", overlap) + } +} + +// ---------- 小工具 ---------- + +func splitSSEFrames(s string) []string { + var out []string + cur := "" + for i := 0; i < len(s); i++ { + cur += string(s[i]) + if s[i] == '\n' && strings.HasSuffix(cur, "\n\n") { + out = append(out, cur) + cur = "" + } + } + if strings.TrimSpace(cur) != "" { + out = append(out, cur) + } + return out +} + +func dataLine(frame string) string { + for _, ln := range strings.Split(frame, "\n") { + if strings.HasPrefix(ln, "data: ") { + return strings.TrimPrefix(ln, "data: ") + } + } + return "" +} + +func truncStr(s string, n int) string { + if len(s) > n { + return s[:n] + "…" + } + return s +} diff --git a/server/internal/sse/manager.go b/server/internal/sse/manager.go index 4bc436b..6044d57 100644 --- a/server/internal/sse/manager.go +++ b/server/internal/sse/manager.go @@ -103,6 +103,30 @@ type Client struct { Res http.ResponseWriter Flusher http.Flusher done chan struct{} + + // ★ writeMu 串行化对 Res 的**每一次**写入。 + // + // 为什么必需(2026-09-28 实测,-race 证实): + // http.ResponseWriter **不是并发安全**的,而本结构原先**一把写锁都没有**。 + // Manager.mu 只护 `clients` map 的**遍历**,遍历期间的 `c.SendWithID` 是并发的 —— + // 任何两条并发请求都会同时向同一个 client 写。 + // + // 生产上会打中的三条路径: + // ① handler/permission.go:412-414 —— `SendToAgent(perm.AgentName, …)` 紧接 + // `SendToUser(user.Username, …)`,两个不同 HTTP 请求(两个 goroutine)命中同一账号; + // ② 任意两条并发邮件:一封投给 B,B 的插件回信进 C 的 handler,而 A 的 + // `notify.Recipients` 还没跑完; + // ③ `heartbeat` 那条 goroutine 每 10s 写一次(见 heartbeatInterval 的注释), + // 与推送撞车的概率随在线时长线性上升。 + // + // 症状:SSE 是 `id: N\nevent: X\ndata: {…}\n\n` 的**文本协议**,两个 Fprintf + // 交错 ⇒ data 的 JSON 被劈成半截 ⇒ 客户端 EventSource 收到坏帧、丢事件。 + // 实测 32 goroutine × 25 帧 = 800 帧,只切出 **459** 帧完整。 + // + // 为什么不能靠上层串行化:推送方有 5 个入口(SendToUser / SendToAgent / + // SendToRecipient / Broadcast / replay),要保证"同一个 client 的所有写互斥", + // 责任只能落在 client 自己身上 —— 那是唯一能覆盖**全部**写者的位置。 + writeMu sync.Mutex } // Manager 管理所有 SSE 客户端连接 @@ -182,10 +206,19 @@ func (m *Manager) AddClient(res http.ResponseWriter, r *http.Request, agentName, // Last-Event-ID 回放:EventSource 断线重连时自带这个头, // 服务端据此把断线期间的事件补上 —— 否则重连后永远看不到那段时间的邮件。 + // + // ★ 持 writeMu 写入(尽管此时**按构造就是单写者**): + // 回放发生在下面 `m.clients[id] = client` **之前** —— 此刻还没有任何 goroutine + // 拿得到这个 client 的指针,所以本身上就是安全的。持锁是为了让「对 Res 的写入 + // 一律经由 writeMu」成为**结构上**的纪律:将来有人把注册提前、或把回放挪到 + // 注册之后(很自然的一个改动),没上锁的版本会**静默**退化成并发写。 + // 锁在这里零成本,而它买的正是「后人改顺序也不会破」这件事。 lastID := r.Header.Get("Last-Event-ID") key := m.bufferKey(userName, agentName) if ring := m.getOrCreateRing(key); ring != nil && lastID != "" { + client.writeMu.Lock() ring.replay(lastID, flusher, res) + client.writeMu.Unlock() } m.mu.Lock() @@ -334,6 +367,8 @@ func (c *Client) Send(eventType string, data interface{}) { return } + c.writeMu.Lock() + defer c.writeMu.Unlock() fmt.Fprintf(c.Res, "event: %s\ndata: %s\n\n", eventType, jsonData) c.Flusher.Flush() } @@ -347,6 +382,8 @@ func (c *Client) SendWithID(id, eventType string, data interface{}) { return } + c.writeMu.Lock() + defer c.writeMu.Unlock() fmt.Fprintf(c.Res, "id: %s\nevent: %s\ndata: %s\n\n", id, eventType, jsonData) c.Flusher.Flush() } @@ -373,8 +410,13 @@ func (m *Manager) heartbeat(client *Client) { return case <-ticker.C: defer func() { recover() }() + // ★ 同一把 writeMu:心跳是本结构里**第三条**写 Res 的路径。 + // 漏了它就等于"推送之间互斥、心跳不参与"—— 而心跳每 10s 一次、 + // 覆盖连接的全部存活期,撞上推送是必然事件(见 writeMu 的注释 ③)。 + client.writeMu.Lock() fmt.Fprintf(client.Res, ": heartbeat\n\n") client.Flusher.Flush() + client.writeMu.Unlock() } } }