From 968b01f26e7ec18a3a50f2f6ca684e70fc72b99c Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sat, 5 Sep 2026 05:53:17 +0800 Subject: [PATCH] =?UTF-8?q?fix(plugins):=20=E4=BF=AE=E5=A4=8D=E9=9A=94?= =?UTF-8?q?=E7=A6=BB=E5=85=A8=E9=87=8F=E6=B5=8B=E8=AF=95=E6=9A=B4=E9=9C=B2?= =?UTF-8?q?=E7=9A=84=E4=B8=A4=E5=A4=84=E6=95=B0=E6=8D=AE=E7=AB=9E=E4=BA=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit -go test ./... -race 全仓复验暴露的 7 处 race、5 个失败测试，全部定位。 两处独立缺陷，互不相关。 ## 缺陷一（remotedevice，8 处 race）：连接写无串行化 WARNING: DATA RACE Read at 0x... by goroutine 28: bufio.(*Writer).Available() / writeFrameHeader / PushData Previous write at 0x... by goroutine 27: bufio.(*Writer).Flush() / writeFrame / handleWS 同一连接的 bufio.Writer 被两条并发路径写： - handleWS 主循环：读到设备帧后回写 hello_ack/bind_ack/pong - PushJSON/PushData：agent→设备的下发路径，可来自任意 goroutine bufio.Writer 不是线程安全的。不加锁就在 WriteByte/Flush 上撞——这 不是理论风险，TestWSPushDataAudio 的异步 PushData 与 handleWS 的 hello_ack 回写并发时被 -race 稳定抓到。 修法：wconn 增加 wmu（sync.Mutex），PushJSON/PushData 拿锁后整条 下发（start + N 个 chunk + end）持锁——设备侧按协议串行聚合，中途 被插帧会破坏协议顺序。handleWS 的 hello_ack/bind_ack/pong 也改走 同一把锁（wsWriteLocked 封装，避免调用方绕过）。设备已离线时不回写。 关键点：不能只锁 Push* 不锁 handleWS——那只是把竞争挪了个位置。 ## 缺陷二（agentcli，1 处 race）：共享读缓冲被并发读写 WARNING: DATA RACE Write at 0x... by goroutine 26: os.File.Read / (*linuxPty).Read / reader Previous read at 0x... by goroutine 25: runtime.slicecopy / readLoop readLoop 创建 buf := make([]byte, ReadBufSize) 传给 reader goroutine （t.session.Read(buf) 持续覆写），自己又在读到结果后 copy(data, buf[:r.n])——同一缓冲被读写并发。Go 的 pty 读走 OS 层 fd，专门在 reader 写下一段时读，跑 -race 稳定复现。 修法：readResult 携带 data field，reader 每次读完后把数据复制进自己 分配的切片再随结果传递，读取与拷贝之间不再共享任何可变状态。原 buf 保留（仍由 reader 独享用于 OS 读），readLoop 不再从其中 copy。 ## 验证 - 两个插件包 -race -count=2 全过 - 全仓 go build / go vet / go test 通过 - 全仓 go test ./... -race：32 包全过，0 DATA RACE，0 FAIL - SDK 冻结 diff = 0 其中 remotedevice 的 TestScreenseeEndToEnd / TestComputeruseEndToEnd / TestClipboardEndToEnd 原本因 race 挂，修后恢复全绿。 --- internal/plugins/agentcli/plugin.go | 64 +++++++++++-------- internal/plugins/remotedevice/registry.go | 78 +++++++++++++++++++---- 2 files changed, 105 insertions(+), 37 deletions(-) diff --git a/internal/plugins/agentcli/plugin.go b/internal/plugins/agentcli/plugin.go index e9e133e..3b38520 100644 --- a/internal/plugins/agentcli/plugin.go +++ b/internal/plugins/agentcli/plugin.go @@ -18,10 +18,10 @@ import ( ) const ( - DefaultTimeout = 5 * time.Minute - ReadBufSize = 4096 - MaxOutputBuffer = 128 * 1024 - DefaultNotifyBytes = 2048 // 积累 2KB 未读输出再通知 + DefaultTimeout = 5 * time.Minute + ReadBufSize = 4096 + MaxOutputBuffer = 128 * 1024 + DefaultNotifyBytes = 2048 // 积累 2KB 未读输出再通知 DefaultNotifyInterval = 2 * time.Second // 同一终端两次通知的最小间隔（兜底） ) @@ -68,12 +68,12 @@ type TerminalSession struct { done chan struct{} // 通知节流字段 - unreadBytes int // 最近一次通知后积累的未读字节数 - lastNotify time.Time // 最近一次通知时间 - lastData time.Time // 最近一次读到的数据时间（用于判定输出停止） - lastFeedback time.Time // 最近一次定时反馈时间 - backoff time.Duration // 输出风暴退避：持续高速输出时通知间隔翻倍 - watch terminalWatch // 该终端的提醒规则 + unreadBytes int // 最近一次通知后积累的未读字节数 + lastNotify time.Time // 最近一次通知时间 + lastData time.Time // 最近一次读到的数据时间（用于判定输出停止） + lastFeedback time.Time // 最近一次定时反馈时间 + backoff time.Duration // 输出风暴退避：持续高速输出时通知间隔翻倍 + watch terminalWatch // 该终端的提醒规则 // 实时画面推流（terminal_output 事件） stream bytes.Buffer // 待推送的增量输出，由 readLoop 每 200ms flush 一次 @@ -226,7 +226,7 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { Description: "创建一个新的交互式终端会话。返回终端 ID，后续通过此 ID 进行读写操作。适用于运行交互式程序如 vim、ssh、top、nano 等。" + "通知模式通过 notify 参数选择（默认 exit）：exit=仅命令执行结束后提醒一次；interval=定时反馈（如 interval=30s 每 30 秒反馈一次状态摘要）；" + "buffer=未读输出积累到指定字节数后提醒（如 buffer=8192）；多个模式用逗号组合（如 interval=30s,buffer=8192）。终端默认 5 分钟后自动关闭，可通过 timeout 参数调整。", - NoMemory: true, + NoMemory: true, Parameters: map[string]interface{}{ "type": "object", "properties": map[string]interface{}{ @@ -283,7 +283,7 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { }) s.RegisterTool("terminal_read", sdk.ToolDef{ - Name: "terminal_read", + Name: "terminal_read", Description: "读取指定终端的输出。mode=new（默认）返回自上次读取以来的新输出并清空缓冲；mode=now 返回终端当前显示的全部屏幕内容（不清空缓冲）。如需持续监控请多次调用。", NoMemory: true, Parameters: map[string]interface{}{ @@ -759,11 +759,11 @@ func (p *Plugin) handleList() (interface{}, error) { defer p.mu.Unlock() type termInfo struct { - ID string `json:"id"` - Command string `json:"command"` - Uptime string `json:"uptime"` + ID string `json:"id"` + Command string `json:"command"` + Uptime string `json:"uptime"` ExpiresIn string `json:"expires_in"` - Running bool `json:"running"` + Running bool `json:"running"` } var terms []termInfo @@ -787,8 +787,8 @@ func (p *Plugin) handleList() (interface{}, error) { } return map[string]interface{}{ - "status": "ok", - "count": len(terms), + "status": "ok", + "count": len(terms), "terminals": terms, }, nil } @@ -797,6 +797,9 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) { defer p.wg.Done() defer close(t.done) + // reader 协程独享这个读缓冲：结果随 readResult 携带， + // readLoop 不再从其中做 copy（见 reader 注释，那是对共享缓冲 + // 的并发读写，-race 实测触发）。 buf := make([]byte, ReadBufSize) pollInterval := 200 * time.Millisecond @@ -816,7 +819,7 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) { t.lastFeedback = now t.mu.Unlock() -// 硬上限：未读输出积累达到该值也通知一次（防大输出静默丢失），频率极低 + // 硬上限：未读输出积累达到该值也通知一次（防大输出静默丢失），频率极低 hardNotifyBytes := 64 * 1024 hardNotifyInterval := 10 * time.Second // 输出停止判定：超过该时长无新数据则视为输出停止 @@ -886,9 +889,7 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) { return } if r.n > 0 { - data := make([]byte, r.n) - copy(data, buf[:r.n]) - t.appendOutput(data) + t.appendOutput(r.data) // 缓冲阈值通知（仅当 agent 显式选择 buffer 模式，或未读积累达到硬上限）。 // 默认模式（仅 exit 提醒）下不随输出流通知，杜绝通知风暴。 @@ -951,15 +952,28 @@ func previewTail(s string, n int) string { } type readResult struct { - n int - err error + n int + data []byte + err error } +// reader 从终端读取输出并通过 channel 交给 readLoop。 +// +// 读到的数据**随结果一起传**而不是复用外层共享的 buf： +// reader 是唯一写 buf 的 goroutine，readLoop 又常在 reader 尚未 +// 写完下一段时就从 buf[:r.n] 做 copy——同一个 shared buf 被并发 +// 读写就是 data race（-race 实测触发）。改为每个结果自带切片后， +// 读与拷贝天然隔离，不再共享可变状态。 func (p *Plugin) reader(t *TerminalSession, buf []byte, ch chan<- readResult) { for { n, err := t.session.Read(buf) + var data []byte + if n > 0 { + data = make([]byte, n) + copy(data, buf[:n]) + } select { - case ch <- readResult{n, err}: + case ch <- readResult{n, data, err}: case <-t.stopCh: return } diff --git a/internal/plugins/remotedevice/registry.go b/internal/plugins/remotedevice/registry.go index bc8db01..6a0fb05 100644 --- a/internal/plugins/remotedevice/registry.go +++ b/internal/plugins/remotedevice/registry.go @@ -37,6 +37,24 @@ type DeviceMeta struct { type wconn struct { deviceID string w *bufio.Writer + // wmu 序列化对该连接 bufio.Writer 的所有写。 + // + // 两个角色会并发写同一连接：handleWS 主循环（读设备帧后的 hello_ack/ + // bind_ack/pong 回写）与 PushJSON/PushData（agent→设备的下发路径，可能 + // 来自任意 goroutine）。bufio.Writer 不是线程安全的，不加锁会在 + // WriteByte/Flush 上产生 data race（生产实测触发）。 + wmu sync.Mutex +} + +// lockWrite 对 wconn 加写锁并返回 writer；调用方必须 defer unlockWrite。 +// 单独写成方法而不是直接暴露字段，避免调用方绕过锁。 +func (c *wconn) lockWrite() *bufio.Writer { + c.wmu.Lock() + return c.w +} + +func (c *wconn) unlockWrite() { + c.wmu.Unlock() } // Registry 是设备接入网关的注册表：管理在线连接、设备元数据。线程安全。 @@ -319,7 +337,9 @@ func (r *Registry) PushJSON(deviceID string, payload map[string]interface{}) err if !ok { return fmt.Errorf("device %s not online", deviceID) } - return writeText(c.w, mustJSON(payload)) + w := c.lockWrite() + defer c.unlockWrite() + return writeText(w, mustJSON(payload)) } // PushCmd 向设备发送命令执行请求。 @@ -348,7 +368,11 @@ func (r *Registry) PushData(deviceID, reqID, kind, mime string, data []byte) err if !ok { return fmt.Errorf("device %s not online", deviceID) } - if err := writeText(c.w, mustJSON(map[string]interface{}{ + // 整条下发（start + N 个 chunk + end）持锁：设备侧按协议串行聚合， + // 若中途被 handleWS 的 hello/pong 插帧会破坏协议顺序。 + w := c.lockWrite() + defer c.unlockWrite() + if err := writeText(w, mustJSON(map[string]interface{}{ "op": "cmd_speech_start", "req_id": reqID, "kind": kind, @@ -363,11 +387,11 @@ func (r *Registry) PushData(deviceID, reqID, kind, mime string, data []byte) err if end > len(data) { end = len(data) } - if err := writeBinary(c.w, data[off:end]); err != nil { + if err := writeBinary(w, data[off:end]); err != nil { return fmt.Errorf("push data chunk: %w", err) } } - if err := writeText(c.w, mustJSON(map[string]interface{}{ + if err := writeText(w, mustJSON(map[string]interface{}{ "op": "cmd_speech_end", "req_id": reqID, })); err != nil { @@ -618,6 +642,26 @@ func (r *Registry) ServeWS(w http.ResponseWriter, req *http.Request) { go r.handleWS(conn, rw) } +// wsWriteLocked 在指定设备连接的写锁保护下执行写回调。 +// +// handleWS 主循环与 Push* 是两条并发写同一 bufio.Writer 的路径， +// 必须共用同一把锁。handleWS 里拿到的是 rw.Writer（与 conns 存储的是 +// 同一个对象），回写前必须经此函数取锁，否则跟 Push* 依然会撞。 +// +// 注意设备已离线（conns 中已删除）时直接报错——设备断开后仍尝试 +// 回写没有意义，还可能在已关闭的 bufio 上写入。 +func (r *Registry) wsWriteLocked(deviceID string, fn func(w *bufio.Writer) error) error { + r.mu.RLock() + c, ok := r.conns[deviceID] + r.mu.RUnlock() + if !ok { + return fmt.Errorf("device %s not online", deviceID) + } + w := c.lockWrite() + defer c.unlockWrite() + return fn(w) +} + func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { defer conn.Close() var curID string @@ -635,7 +679,9 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { payload, isClose, opcode, err := readFrame(rw.Reader) if err != nil { if err == errPing { - if werr := writePong(rw.Writer); werr != nil { + // pong 也走写锁：它可能在 Push* 持锁推送大块数据时到达。 + err := r.wsWriteLocked(curID, writePong) + if err != nil { return } continue @@ -679,11 +725,13 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { r.mu.Lock() r.conns[meta.DeviceID] = &wconn{deviceID: meta.DeviceID, w: rw.Writer} r.mu.Unlock() - if err := writeText(rw.Writer, mustJSON(map[string]interface{}{ - "op": "hello_ack", - "device": meta.DeviceID, - "online": true, - })); err != nil { + if err := r.wsWriteLocked(meta.DeviceID, func(w *bufio.Writer) error { + return writeText(w, mustJSON(map[string]interface{}{ + "op": "hello_ack", + "device": meta.DeviceID, + "online": true, + })) + }); err != nil { return } case "bind": @@ -694,11 +742,17 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { // 默认不授权：bind 仅验证 token + 登记设备；授权完全由用户手动 // （GUI 设备页 / REST /api/v1/device/auth）控制，绝不自动授权。 } - if err := writeText(rw.Writer, mustJSON(map[string]interface{}{"op": "bind_ack", "ok": true})); err != nil { + err := r.wsWriteLocked(curID, func(w *bufio.Writer) error { + return writeText(w, mustJSON(map[string]interface{}{"op": "bind_ack", "ok": true})) + }) + if err != nil { return } } else { - if err := writeText(rw.Writer, mustJSON(map[string]interface{}{"op": "bind_ack", "ok": false, "error": "bad token"})); err != nil { + err := r.wsWriteLocked(curID, func(w *bufio.Writer) error { + return writeText(w, mustJSON(map[string]interface{}{"op": "bind_ack", "ok": false, "error": "bad token"})) + }) + if err != nil { return } }