jianf/HomeAgent
1
0
Fork 0
mirror of https://gitcode.com/JianFeeeee/HomeAgent.git synced 2026-09-21 01:18:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(plugins): 修复隔离全量测试暴露的两处数据竞争

Browse Source 
-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 挂，修后恢复全绿。
This commit is contained in:
JianFeeeee
2026-09-05 05:53:17 +08:00
parent 98dcb2556c
commit 968b01f26e
2 changed files with 105 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
internal/plugins/agentcli/plugin.go
View File

@ -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
}

78
internal/plugins/remotedevice/registry.go
View File

@ -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
}
}