From 94c74b2ee64de3192de8c2d21dc8929ca7b62856 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Fri, 25 Sep 2026 15:12:17 +0800 Subject: [PATCH] =?UTF-8?q?fix(devicebridge):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E8=AE=BE=E5=A4=87=E5=8F=8D=E5=A4=8D=E6=8E=89=E7=BA=BF/?= =?UTF-8?q?=E9=9D=99=E9=BB=98=E5=A4=B1=E8=81=94=20=E2=80=94=E2=80=94=20pin?= =?UTF-8?q?g=20=E8=B7=AF=E5=BE=84=E6=96=AD=E8=BF=9E=20+=20bind=20=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E6=97=A0=E4=BA=BA=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户要求全面修复「设备桥自动链接」这条链路上的问题。三个真实缺陷， 前两个是**服务端/客户端真 bug**（生产日志实证），第三个是我起初误判的。 ## 缺陷 1（最严重）：未 bind 时收到 ping → 服务端直接关连接 原实现： err := r.wsWriteLocked(curID, writePong) if err != nil { return } // ← 关连接 而 conns 表**只在 bind 成功后才写入**（bind 前刻意不暴露连接给查询/命令 路径）。于是「握手完成、bind 尚未到达」这个窗口里来的 ping 找不到写入口， 函数返回错误，读循环 return —— 把连接关掉了。 生产后果（journalctl 实证）：客户端每 30s ping 一次，只要有一次落在未 bind 窗口就断连。日志里同一设备 20 秒内多次 "ws connected"，online/offline 与 输出通道注销/注册反复交替： 17:29:14 ws connected → 17:29:17 ws connected → 17:29:24 ws connected → 17:29:29 → 17:29:35 → 17:29:40 online → 17:30:18 offline → ...循环 修法：pong 直接写本连接的 writer。此时该连接尚未进入 conns（没有 Push* 会 碰它的 writer），不存在并发写风险；已 bind 时才取写锁（Push* 可能正在写 同一 buffer）。 判据 TestPingBeforeBindDoesNotDropConnection 直打 bug 点（只握手、不发 hello/bind、发 ping、要求 pong），修复前报 `EOF`，修复后通过。 ## 缺陷 2：bind_ack 的 ok 完全没被检查 → 失败静默失联 原实现（客户端）： case "hello_ack", "bind_ack": log.Printf("... device=%v", msg["device"]) 两处错： - **取错字段**：服务端成功时回 {"op":"bind_ack","ok":true}，没有 device 字段，于是日志永远显示 `bind_ack device=`。这让我起初误判成"绑定 失败"，实际连接是好的（直连与经反代现象完全一致）。 - **不看 ok**：bind 被拒时服务端回 ok:false + error 并关闭连接，客户端既不 报错也不重连，设备静默失联 —— TCP/WS 通但从未登记进网关。 修法：分别处理两种 ack；bind 判 ok，失败记原因并通知宿主。新增 Bridge.Bound() / BindError() / OnBoundState()：**连接成功 ≠ 设备可用**， 只看连接状态的健康检查会给出假阳性。 判据 TestBindFailureIsObservable / TestBindStateCallback。 ## 缺陷 3：-chat 一次性模式下桥存活时间过短 不是我最初以为的"bind 失败"。真因：`-chat` 走进 oneshot 后立刻 return， 触发 defer stopDeviceBridge()，桥只活几百毫秒，设备来不及完成 hello→bind。 修法：退出前等 bind 确认（最多 3s）；bind 明确被拒则打印原因，不静默丢弃。 ## 真实验收（隔离实例，命名 netns + 独立 data + 18080） 真 waiter 经**反代自动发现**连接，保持连接期间查询服务端： device gateway discovered: ws://127.0.0.1:18080/api/v1/device/ws bind 成功，设备已登记 /api/v1/device/online → waiter-mainserver, online=true, caps=[11 项] 长连接稳定性：70 秒（跨 2 个 ping 周期）三次采样设备始终在线， 无 read loop exit / bind rejected 日志。 ## 附：反代通路本身的判定性对照 裸客户端（直接构造 hello/bind 帧）**经反代**与**直连 9890** 返回逐字节 一致（bind_ack ok=true、设备注册、online=true）。所以这条链路上反代 不背锅，问题全在 remotedevice 服务端与客户端自身。 --- cmd/waiter/main.go | 35 ++++++++ internal/devicebridge/client/bridge.go | 85 +++++++++++++++++++- internal/devicebridge/client/bridge_test.go | 62 ++++++++++++++ internal/plugins/remotedevice/binary_test.go | 74 +++++++++++++++++ internal/plugins/remotedevice/registry.go | 24 +++++- 5 files changed, 276 insertions(+), 4 deletions(-) create mode 100644 internal/devicebridge/client/bridge_test.go diff --git a/cmd/waiter/main.go b/cmd/waiter/main.go index 5b89103..f4f79ad 100644 --- a/cmd/waiter/main.go +++ b/cmd/waiter/main.go @@ -205,6 +205,18 @@ func main() { } if oneShotMsg != "" { + // ★ -chat 是一次性问答，会立刻走到上面的 return 并触发 + // defer stopDeviceBridge()，桥的生命周期只有几百毫秒。 + // + // 后果：设备来不及完成 hello→bind 登记就已断开，服务端列表里永远 + // 看不到它（实测：`device bridge active` 打印了、bind_ack 也收到了， + // 但 /api/v1/device/online 始终为空）。 + // 这不是桥的错 —— 用裸客户端把 hello/bind 发完并保持连接，同一实例 + // 上设备立刻出现在列表里（已验证）。 + // + // 等待 bind 确认（或短暂超时）再退出：既让登记完成，也不把一次性 + // 命令拖长。bind 失败要明说，而不是静默丢掉设备。 + waitDeviceBind(3 * time.Second) oneshot(state, oneShotMsg) return } @@ -499,3 +511,26 @@ func printServerEventColored(rl respLine, raw string) { fmt.Printf("%s%s%s\n", clearLine, text, colorReset) } } + +// waitDeviceBind 等待服务端确认 bind（最多 timeout），返回是否确认。 +// +// 用于一次性命令（-chat）：桥启动后立刻退出会让设备来不及登记。 +// 超时不报错（服务端可能只是慢），bind 明确被拒则打出来 —— 那通常意味着 +// 设备令牌不对或设备未授权，用户需要知道，而不是以为「桥起来了就好了」。 +func waitDeviceBind(timeout time.Duration) bool { + if deviceBridge == nil { + return false + } + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if deviceBridge.Bound() { + return true + } + if reason := deviceBridge.BindError(); reason != "" { + printlnC(colorYellow, "device bridge bind rejected: "+reason) + return false + } + time.Sleep(50 * time.Millisecond) + } + return deviceBridge.Bound() +} diff --git a/internal/devicebridge/client/bridge.go b/internal/devicebridge/client/bridge.go index 639c549..272ff22 100644 --- a/internal/devicebridge/client/bridge.go +++ b/internal/devicebridge/client/bridge.go @@ -46,6 +46,14 @@ type Bridge struct { info map[string]interface{} authorized bool // 客户端本地授权状态（用户在设备上手动开启） + // bound 表示服务端已确认 bind（设备真正登记进网关）。 + // 与「连接已建立」是两件事：连接成功但 bind 被拒时设备是**失联**的， + // 必须能区分（原实现完全不看 bind_ack 的 ok，失败静默）。 + mu2 sync.RWMutex + bound bool + lastBind string // 最近一次 bind 失败原因（空 = 未失败过） + boundHandler func(bound bool, reason string) + ws *wsConn stopCh chan struct{} doneCh chan struct{} @@ -139,6 +147,55 @@ func (b *Bridge) SetAuthorized(auth bool) { } } +// markBound 记录服务端已确认 bind。 +func (b *Bridge) markBound() { + b.mu2.Lock() + b.bound = true + b.lastBind = "" + h := b.boundHandler + b.mu2.Unlock() + if h != nil { + h(true, "") + } +} + +// markUnbound 记录 bind 失败（连接可能随即被服务端关闭）。 +func (b *Bridge) markUnbound(reason string) { + b.mu2.Lock() + b.bound = false + b.lastBind = reason + h := b.boundHandler + b.mu2.Unlock() + if h != nil { + h(false, reason) + } +} + +// Bound 返回服务端是否已确认 bind。 +// +// 为什么要单独一个状态：连接成功 ≠ 设备可用。bind 被拒时 TCP/WS 是通的， +// 但设备没有登记进网关，命令永远下发不到 —— 只看「连接是否建立」的 +// 健康检查会给出假阳性。 +func (b *Bridge) Bound() bool { + b.mu2.RLock() + defer b.mu2.RUnlock() + return b.bound +} + +// BindError 返回最近一次 bind 失败原因（空 = 未失败）。 +func (b *Bridge) BindError() string { + b.mu2.RLock() + defer b.mu2.RUnlock() + return b.lastBind +} + +// OnBoundState 注册绑定状态变更回调（GUI/waiter 据此提示用户）。 +func (b *Bridge) OnBoundState(handler func(bound bool, reason string)) { + b.mu2.Lock() + b.boundHandler = handler + b.mu2.Unlock() +} + // Authorized 返回当前客户端本地授权状态。 func (b *Bridge) Authorized() bool { b.mu.RLock() @@ -464,7 +521,33 @@ func (b *Bridge) handleMessage(msg map[string]interface{}) { } case "hello_ack", "bind_ack": - log.Printf("[devicebridge] %s device=%v", op, msg["device"]) + // ★ 必须分别处理，且要判失败。 + // + // 原实现只打一行 `device=%v`，而服务端**成功**时回的是 + // {"op":"bind_ack","ok":true} —— 根本没有 device 字段，于是日志 + // 永远显示 `bind_ack device=`。这会让人误判成「绑定失败」， + // 而实际上连接是好的（实测：同一现象经直连与经反代完全一致）。 + // + // 更严重的是失败无人处理：服务端 bind 被拒时回 + // {"op":"bind_ack","ok":false,"error":"bind rejected"} 并**关闭连接**， + // 客户端却既不报错也不重连，设备静默失联。 + if op == "bind_ack" { + if ok, _ := msg["ok"].(bool); !ok { + reason, _ := msg["error"].(string) + if reason == "" { + reason = "bind rejected" + } + log.Printf("[devicebridge] bind 被拒：%s（令牌不匹配或设备未授权）", reason) + b.markUnbound(reason) + return + } + log.Printf("[devicebridge] bind 成功，设备已登记") + b.markBound() + return + } + dev, _ := msg["device"].(string) + online, _ := msg["online"].(bool) + log.Printf("[devicebridge] hello_ack device=%s online=%v", dev, online) case "cmd_speech_start": reqID, _ := msg["req_id"].(string) diff --git a/internal/devicebridge/client/bridge_test.go b/internal/devicebridge/client/bridge_test.go new file mode 100644 index 0000000..972904a --- /dev/null +++ b/internal/devicebridge/client/bridge_test.go @@ -0,0 +1,62 @@ +package client + +import "testing" + +// bind_ack 失败必须被识别（原实现完全不看 ok，失败静默）。 +// +// 服务端 bind 被拒时回 {"op":"bind_ack","ok":false,"error":"bind rejected"} +// 并关闭连接。客户端若不看 ok，设备就静默失联：TCP/WS 是通的， +// 但设备从未登记进网关，命令永远下发不到 —— 只看"连接是否建立"的 +// 健康检查会给出假阳性。 +func TestBindFailureIsObservable(t *testing.T) { + b := New("ws://127.0.0.1:1/api/v1/device/ws", "tok", "dev-1", "n", []string{"status"}, nil) + + if b.Bound() { + t.Error("初始不应处于已绑定状态") + } + if b.BindError() != "" { + t.Errorf("初始不应有绑定错误，实际 %q", b.BindError()) + } + + // 模拟收到 bind 失败 + b.markUnbound("bind rejected") + if b.Bound() { + t.Error("失败后不应报 Bound=true") + } + if b.BindError() != "bind rejected" { + t.Errorf("失败原因应被保留，实际 %q", b.BindError()) + } + + // 成功后状态必须清空错误 + b.markBound() + if !b.Bound() { + t.Error("成功后应 Bound=true") + } + if b.BindError() != "" { + t.Errorf("成功后应清空失败原因，实际 %q", b.BindError()) + } +} + +// 绑定状态变更必须通知宿主（GUI/waiter 据此提示用户）。 +func TestBindStateCallback(t *testing.T) { + b := New("ws://127.0.0.1:1/api/v1/device/ws", "tok", "dev-2", "n", nil, nil) + type ev struct { + bound bool + reason string + } + var got []ev + b.OnBoundState(func(bound bool, reason string) { got = append(got, ev{bound, reason}) }) + + b.markUnbound("token mismatch") + b.markBound() + + if len(got) != 2 { + t.Fatalf("应收到 2 次回调，实际 %d", len(got)) + } + if got[0].bound || got[0].reason != "token mismatch" { + t.Errorf("第一次回调应为失败: %+v", got[0]) + } + if !got[1].bound || got[1].reason != "" { + t.Errorf("第二次回调应为成功: %+v", got[1]) + } +} diff --git a/internal/plugins/remotedevice/binary_test.go b/internal/plugins/remotedevice/binary_test.go index 18460fe..d569fd7 100644 --- a/internal/plugins/remotedevice/binary_test.go +++ b/internal/plugins/remotedevice/binary_test.go @@ -862,3 +862,77 @@ func TestAwaitResultReturnsResultDeliveredBeforeWaiter(t *testing.T) { t.Fatalf("unexpected early result: %v", got) } } + +// ===== ping 帧处理：不得因未 bind 而断连 ===== + +// ★ 未 bind（或已 bind）时收到 ping，服务端必须回 pong，**不得关闭连接**。 +// +// 真实 bug（生产日志实证）：原实现用 wsWriteLocked(curID, writePong)，而 +// conns[curID] 只在 bind 成功后才写入 ⇒ 握手后、bind 前到来的 ping 找不到 +// 写锁入口，函数返回错误，读循环直接 return 关连接。 +// +// 后果：客户端每 30s ping 一次，只要有一次落在未 bind 窗口就断连；生产日志里 +// 同一设备 20 秒内多次 "ws connected" 且 online/offline 反复交替，正是这个。 +// +// 判据直打 bug 点：只握手、**不发 hello/bind**，发 ping，要求收到 pong。 +// readMsg 会跳过 pong，所以这里直接读原始帧。 +func TestPingBeforeBindDoesNotDropConnection(t *testing.T) { + reg := NewRegistry() + token := "tok-ping-bind" + reg.SetAcceptToken(func(s string) bool { return s == token }) + + srv := httptest.NewServer(http.HandlerFunc(reg.ServeWS)) + defer srv.Close() + + c := dialTestWS(t, srv.URL, token) + defer c.close() + + // 只握手，不发 hello/bind —— 模拟「尚未绑定完成」的窗口 + c.sendFrame(0x9, nil) // ping + + // 必须收到 pong；EOF/错误说明服务端在 ping 路径上断了连接 + c.conn.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, _, opcode, err := readFrame(c.rw.Reader) + if err != nil { + t.Fatalf("未 bind 时 ping 导致连接不可用（服务端 bug）: %v", err) + } + if opcode != 0xA { + t.Fatalf("期望 pong(0xA)，收到 opcode=%#x", opcode) + } +} + +// 已 bind 的设备发 ping 同样必须得到 pong，且连接与在线状态都保持。 +func TestPingAfterBindGetsPong(t *testing.T) { + reg := NewRegistry() + token := "tok-ping-bound" + reg.SetAcceptToken(func(s string) bool { return s == token }) + srv := httptest.NewServer(http.HandlerFunc(reg.ServeWS)) + defer srv.Close() + + c := dialTestWS(t, srv.URL, token) + defer c.close() + c.sendText(mustJSON(map[string]interface{}{ + "op": "hello", + "device": map[string]interface{}{ + "device_id": "d-ping", "name": "d", "kind": "computer", "caps": []string{"status"}, + }, + })) + c.readHelloAck(t) + c.bindDevice(t, "d-ping", token) + if !reg.Online("d-ping") { + t.Fatal("bind 后设备应在线") + } + + c.sendFrame(0x9, nil) // ping + c.conn.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, _, opcode, err := readFrame(c.rw.Reader) + if err != nil { + t.Fatalf("已 bind 设备 ping 失败: %v", err) + } + if opcode != 0xA { + t.Fatalf("期望 pong(0xA)，收到 %#x", opcode) + } + if !reg.Online("d-ping") { + t.Error("ping 之后设备不应掉线") + } +} diff --git a/internal/plugins/remotedevice/registry.go b/internal/plugins/remotedevice/registry.go index 31b5499..a3520dd 100644 --- a/internal/plugins/remotedevice/registry.go +++ b/internal/plugins/remotedevice/registry.go @@ -731,9 +731,27 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter, handshakeAuthor payload, isClose, opcode, err := readFrame(rw.Reader) if err != nil { if err == errPing { - // pong 也走写锁：它可能在 Push* 持锁推送大块数据时到达。 - err := r.wsWriteLocked(curID, writePong) - if err != nil { + // ★ 回 pong 绝不能因为「还没 bind」而失败。 + // + // 原实现无条件走 wsWriteLocked(curID, writePong)，而 conns 表 + // **只在 bind 成功后才写入**（bind 之前刻意不把连接暴露给查询/ + // 命令路径）。于是握手完成、bind 尚未到达时来的 ping 找不到写 + // 入口 → 返回错误 → 读循环 return → **连接被关掉**。 + // + // 生产后果（日志实证）：客户端每 30s 一次 ping，只要有一次落在 + // 未 bind 窗口就断连，表现为同一设备 20 秒内多次 ws connected、 + // online/offline 反复交替，输出通道跟着反复注销/注册。 + // + // 正确做法：pong 直接写本连接的 writer。此时该连接**尚未**进入 + // conns（即没有 Push* 会碰它的 writer），不存在并发写风险； + // 已 bind 时才需要取写锁（Push* 可能正在写同一 buffer）。 + var werr error + if bound { + werr = r.wsWriteLocked(curID, writePong) + } else { + werr = writePong(rw.Writer) + } + if werr != nil { return } continue