修局域网/中转对比基准：它自己有两个 bug，害产品背了黑锅

这个基准一直是红的（中转那条路报「连接阶段失败: timed out」），而且报告里写着"这是中转链路的产品 bug"。**查下来不是产品的问题。** == bug 1：两条路用了同一个配对码 == 基准为了排除互相干扰，给两条路**各起一台全新被控端**（设计是对的），但两台都用了写死的 code="246813"。中转有「一个配对码只属于一台设备」的规则：第 1 台（跑局域网）注册成功 -> stop_agent 杀掉进程，但**中转还以为它在线**（last_seen 没过期）第 2 台（跑中转）用同一个码注册 -> **被拒** 中转日志：配对码 246813 已被 ... 占用，拒绝 ... 而 start_agent 只检查"有没有设备在线" —— 第 1 台僵尸还挂着，于是**以为新的一台登记成功了** -> 没有任何进程在回包 -> 超时对照实验（新增 _build/verify_code_reuse.py）：场景 A 同一个码 -> **不通** 场景 B 各用不同码 -> 通修法：两条路各用不同码；start_agent 的判据从"有没有在线设备" 改成"有没有**配对码正好是这一台要用的码**的在线设备"。 == bug 2：attach 用的还是旧码（我第一版补丁也漏了这处）== relay_connect() 里 `client.attach(code)` 用的还是局域网那个码 —— attach 到一台已经死掉、但中转以为在线的设备上，pair 自然没人回。修成 `attach(relay_code)`。 == 顺带更新一段过时注释 == 原文写着「当前版本经中转只稳定支撑前 1~2 次交互……这是中转链路的产品 bug」，并据此把中转的轮数调小、超时调短。那个 bug **已经修好了**（根因：一条 HTTP/1.1 连接被长轮询和发命令两个线程同时用，撞状态机抛 CannotSendRequest("Idle")，异常冒到 _client_loop 外面把 client-relay 线程打死）。注释不更新会误导后人以为产品还有这个问题。 == 修好之后的实测（本地中转，30 次/路）== 局域网握手 16ms 成功 30/30 (100%) 中位 0ms P90 0ms 最大 1ms 中转握手 46ms 成功 30/30 (100%) 中位 15ms P90 29ms 最大 32ms 这是本地中转，量的是**协议本身的开销**（长轮询往返 vs 直连 TCP）；真实跨网络那条腿的 1.0~1.5 秒来自隧道和 4G，是另一回事。结论：局域网确实是"毫秒级、不耗流量"的那条根本路径 —— 中位 0ms vs 15ms，而且不走任何外网。踩坑记录：这两处都是"看起来像产品 bug、实际是测试自己的 bug"。先怀疑测量工具，比先怀疑产品更省时间 —— 上一次（管理页缓存头）也是我拿 curl -I 测 GET 白查了半天。
2026-10-18 16:00:43 +00:00 · 2026-10-07 18:20:57 +08:00
parent aac43e559d
commit cc5891e6d0
16 changed files with 1088 additions and 16 deletions
@@ -0,0 +1,225 @@
+# -*- coding: utf-8 -*-
+r'''验证假设：bench_lan_vs_relay 中转那半失败，是因为两条路
+用了**同一个配对码**，第二台被控端注册被"一码一机"规则拒掉，
+而 start_agent 看到上一台还挂在线就以为成功了。
+
+做法：完全照 bench 的流程走两遍（起 agent1 -> 停 -> 起 agent2），
+第一遍用同一个码，第二遍用不同的码，看结果差异。
+'''
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+WINDOWS = ROOT / "windows"
+UA = {"User-Agent": "LanRemote/1.8.2"}
+
+
+def free_port() -> int:
+    s = socket.socket()
+    s.bind(("127.0.0.1", 0))
+    p = s.getsockname()[1]
+    s.close()
+    return p
+
+
+def get(base, path, timeout=6.0):
+    req = urllib.request.Request(base + path, headers=UA)
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return resp.status, resp.read()
+    except Exception:                                     # noqa: BLE001
+        return 0, b""
+
+
+def post(base, path, body=b"", headers=None, timeout=20.0):
+    h = dict(UA)
+    h.update(headers or {})
+    req = urllib.request.Request(base + path, data=body, headers=h,
+                                 method="POST")
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return resp.status, resp.read()
+    except urllib.error.HTTPError as exc:
+        return exc.code, exc.read()
+    except Exception as exc:                              # noqa: BLE001
+        return 0, str(exc).encode()
+
+
+def main() -> int:
+    work = Path(tempfile.mkdtemp(prefix="codechk_"))
+    relay_port = free_port()
+    agent_port = free_port()
+    base = "http://127.0.0.1:%d" % relay_port
+    relay_proc = None
+    procs = []
+    try:
+        sys.path.insert(0, str(WINDOWS))
+        from lanremote.config import Store
+
+        rdata = work / "r"
+        rdata.mkdir(parents=True, exist_ok=True)
+        env = dict(os.environ)
+        env.update({"RELAY_PORT": str(relay_port), "RELAY_BIND": "127.0.0.1",
+                    "RELAY_DATA": str(rdata), "RELAY_POLL": "2",
+                    "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"})
+        rlog = open(work / "relay.log", "wb")
+        relay_proc = subprocess.Popen(
+            [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
+            stdout=rlog, stderr=subprocess.STDOUT)
+        for _ in range(60):
+            time.sleep(0.5)
+            if get(base, "/v1/health")[0] == 200:
+                break
+
+        def start_agent(code, tag):
+            d = work / tag
+            d.mkdir(parents=True, exist_ok=True)
+            st = Store(d)
+            st.set_pairing_code(code)
+            st.set("setup_done", True)
+            st.set("allow_new_pairing", True)
+            st.set("close_pair_after_first", False)
+            st.set("server_name", "验码电脑")
+            st.set("relay_url", base)
+            st.set("port", agent_port)
+            e = dict(os.environ)
+            e.update({"YKXMZ_DATA": str(d),
+                      "YKXMZ_LOCK_PORT": str(free_port()),
+                      "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"})
+            p = subprocess.Popen(
+                [sys.executable, "-u", str(WINDOWS / "run_agent.py"),
+                 "--silent", "--port", str(agent_port)],
+                cwd=str(WINDOWS), env=e,
+                stdout=open(d / "out.log", "wb"), stderr=subprocess.STDOUT)
+            procs.append(p)
+            online = False
+            for _ in range(30):
+                time.sleep(1.0)
+                s, b = get(base, "/v1/health", timeout=4)
+                if s == 200:
+                    snap = json.loads(b.decode())
+                    if any(x.get("online") for x in snap.get("devices", [])):
+                        online = True
+                        break
+            return p, online, snap if s == 200 else {}
+
+        def stop(p):
+            try:
+                p.terminate()
+                p.wait(timeout=6)
+            except Exception:                             # noqa: BLE001
+                try:
+                    p.kill()
+                except Exception:                         # noqa: BLE001
+                    pass
+
+        def relay_try(code, label):
+            """attach + pair，看能不能拿到回复。"""
+            s, b = post(base, "/v1/client/attach", json.dumps(
+                {"code": code, "name": "验码手表"}).encode(),
+                {"Content-Type": "application/json"})
+            att = json.loads(b.decode())
+            if not att.get("ok"):
+                print("    %s attach 就失败: %s" % (label, att.get("error")))
+                return False
+            sess = att.get("session", "")
+            post(base, "/v1/client/send",
+                 (json.dumps({"cmd": "pair", "deviceId": "x",
+                              "code": code}) + "\n").encode(),
+                 {"X-Session": sess})
+            buf = b""
+            deadline = time.time() + 12
+            while time.time() < deadline:
+                _s, chunk = post(base, "/v1/client/poll", b"",
+                                 {"X-Session": sess}, 6.0)
+                buf += chunk
+                while b"\n" in buf:
+                    line, _, buf = buf.partition(b"\n")
+                    if not line.strip():
+                        continue
+                    msg = json.loads(line.decode())
+                    if "event" in msg:
+                        continue
+                    print("    %s pair 成功: %s"
+                          % (label, json.dumps(msg, ensure_ascii=False)[:80]))
+                    return True
+            print("    %s pair **超时**（没人回包）" % label)
+            return False
+
+        # ============ 场景 A：两条路用同一个码（bench 现在的做法）============
+        print("════════ 场景 A：两条路用同一个配对码（复刻 bench 现状）════════")
+        CODE = "246813"
+        p1, on1, snap1 = start_agent(CODE, "a1")
+        print("  第 1 台（跑局域网）: 登记=%s  设备数=%d"
+              % (on1, len(snap1.get("devices", []))))
+        stop(p1)
+        print("  已停掉第 1 台")
+        time.sleep(2)
+        p2, on2, snap2 = start_agent(CODE, "a2")
+        print("  第 2 台（跑中转）: start_agent 认为登记=%s" % on2)
+        for d in snap2.get("devices", []):
+            print("     中转上看到的设备: %s  码 %s  在线=%s  空闲=%ss"
+                  % (d["name"], d["code"], d["online"], d["idleSeconds"]))
+        okA = relay_try(CODE, "场景A")
+        stop(p2)
+
+        print()
+        print("  ── 中转日志里有没有『码被占用』──")
+        rlog.flush()
+        for line in (work / "relay.log").read_text(
+                encoding="utf-8", errors="replace").splitlines():
+            if "占用" in line or "拒绝" in line:
+                print("    " + line[:110])
+
+        # ============ 场景 B：每条路用不同的码 ============
+        print()
+        print("════════ 场景 B：两条路各用不同的配对码（修法）════════")
+        work2 = work / "b"
+        CODE1 = "111222"
+        CODE2 = "333444"
+        p3, on3, _ = start_agent(CODE1, "b1")
+        print("  第 1 台: 登记=%s" % on3)
+        stop(p3)
+        time.sleep(2)
+        p4, on4, snap4 = start_agent(CODE2, "b2")
+        print("  第 2 台: 登记=%s" % on4)
+        okB = relay_try(CODE2, "场景B")
+        stop(p4)
+
+        print()
+        print("════════ 结论 ════════")
+        print("  场景 A（同一个码）: %s" % ("通" if okA else "**不通**"))
+        print("  场景 B（不同码）  : %s" % ("通" if okB else "**不通**"))
+        if (not okA) and okB:
+            print("  → 假设成立：bench 失败是因为两条路用了同一个配对码，")
+            print("     第 2 台被「一码一机」规则拒掉，而 start_agent")
+            print("     看到第 1 台还挂在线就以为成功了。**产品没问题。**")
+        return 0
+    finally:
+        for p in procs:
+            try:
+                p.terminate()
+            except Exception:                             # noqa: BLE001
+                pass
+        if relay_proc:
+            try:
+                relay_proc.terminate()
+                relay_proc.wait(timeout=6)
+            except Exception:                             # noqa: BLE001
+                pass
+        shutil.rmtree(work, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    sys.exit(main())