feat(server): app 窗口租约, 全部窗口关闭自动停 EtherCAT 循环

PDO 循环起来后 native 侧一直跑, 其 --parent-pid 自清理只覆盖"本服务拉起的实例死掉"; 浏览器模式关窗后本服务还在、窗口模式复用现成 ecat_native (parent 是别人)这两种常见离开方式都够不着 —— 从站卡在 OP、自旋节拍白占一个核。现由每个 app 页面与页面服务器的 WebSocket 租约兜底: - esp32_server.py: 新增 GET /ws/app-lease(手写 RFC6455 握手 + 帧泵)。连接计数归零后隔 3s GRACE 调后端 /api/op/stop 停循环退回 INIT; 断开感知靠 TCP FIN / Close 帧回显而非心跳(后台标签页节流不误停); 代理透传时 _note_ecat_loop 记账"循环是否本服务跑起来的", 只停自己确认过的; /api/shutdown 与 serve_forever 退出路径同步补停(定时器随进程死亡)。 - js/common/app-lease.js: 租约客户端, 3~30s 退避重连, file:// 静默跳过。 - 接入 tabs.html(外壳持有, 内部切标签不断连)与 pages/motor.html(单独打开时页面自持)。 - 测试: scripts/test_app_lease.py(裸 TCP 全路径, 假 _ecat_post) + scripts/test_app_lease_e2e.mjs(真 Edge headless + 假后端 8091, 不碰真总线)。README 补设计动机与要点。
2026-09-05 19:58:25 +08:00
parent c189a8563e
commit 99fb9bcb4a
7 changed files with 636 additions and 0 deletions
@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""test_app_lease.py — 验证 "app 全部窗口关闭 → 自动停 EtherCAT 循环" 链路。
+
+不打真实 EtherCAT 硬件: 导入 esp32_server 后把 _ecat_post/_ecat_find_port
+换成假实现, 其余(WS 握手、租约计数、GRACE 吸收、代理记账)全部走真实代码。
+
+用法: python scripts/test_app_lease.py
+"""
+import base64
+import hashlib
+import os
+import socket
+import sys
+import threading
+import time
+import urllib.request
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import esp32_server as srv  # noqa: E402
+
+WS_MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+calls = []            # 假 _ecat_post 记录到的请求路径
+calls_lock = threading.Lock()
+
+
+def fake_post(port, path, body, timeout):
+    with calls_lock:
+        calls.append(path)
+    # 带空格的 "ok": true —— 故意与真 native 的紧凑格式不同,
+    # 锁死"代理记账不能靠子串匹配"这条回归(e2e 里就是这么炸过一次)。
+    return 200, b'{"ok": true, "log": ["fake"]}'
+
+
+def ws_connect(port):
+    """打开一条真实 TCP 上的 WebSocket, 校验 101 握手, 返回 socket。"""
+    s = socket.create_connection(("127.0.0.1", port))
+    key = base64.b64encode(os.urandom(16)).decode()
+    s.sendall(("GET /ws/app-lease HTTP/1.1\r\n"
+               f"Host: 127.0.0.1:{port}\r\n"
+               "Upgrade: websocket\r\nConnection: Upgrade\r\n"
+               f"Sec-WebSocket-Key: {key}\r\n"
+               "Sec-WebSocket-Version: 13\r\n\r\n").encode("ascii"))
+    head = b""
+    while b"\r\n\r\n" not in head:
+        chunk = s.recv(4096)
+        if not chunk:
+            raise AssertionError("握手被直接关闭")
+        head += chunk
+    text = head.decode("ascii", "replace")
+    assert "101" in text.splitlines()[0], f"握手没回 101: {text[:120]}"
+    expect = base64.b64encode(
+        hashlib.sha1((key + WS_MAGIC).encode()).digest()).decode()
+    assert expect in text, "Sec-WebSocket-Accept 不对"
+    return s
+
+
+def ws_graceful_close(s):
+    """浏览器关 WS 的真实行为: 先发掩码 Close 帧, 等服务端回显后才拆 TCP。
+    服务端若不回 Close(RFC 6455 要求), 这条链路的 FIN 永远到不了 → 租约不释放。"""
+    s.sendall(b"\x88\x80" + b"\x00\x00\x00\x00")      # FIN+Close, masked, len 0
+    s.settimeout(3.0)
+    echo = s.recv(2)
+    assert echo == b"\x88\x00", f"Close 回显不对: {echo!r}"
+    s.close()
+
+
+def post_json(port, path):
+    req = urllib.request.Request(
+        f"http://127.0.0.1:{port}{path}", data=b"{}",
+        headers={"Content-Type": "application/json"}, method="POST")
+    with urllib.request.urlopen(req, timeout=10) as r:
+        return r.status, r.read()
+
+
+def flag():
+    with srv.g_app_lock:
+        return srv.g_ecat_motor_live
+
+
+def wait_flag(want, timeout=8.0):
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if flag() == want:
+            return True
+        time.sleep(0.05)
+    return False
+
+
+def main():
+    # 加速测试: GRACE 从 3s 缩到 0.3s(重试轮次间隔同用该值, 一并缩短无妨)
+    srv.APP_LEASE_GRACE_S = 0.3
+    srv._ecat_post = fake_post
+    srv._ecat_find_port = lambda: 9999
+
+    # 找空闲端口起真服务
+    with socket.socket() as probe:
+        probe.bind(("127.0.0.1", 0))
+        port = probe.getsockname()[1]
+    httpd, port = srv._bind_server(port)
+    assert httpd is not None, "起不了测试服务"
+    threading.Thread(target=httpd.serve_forever, daemon=True).start()
+
+    # 1) 非 WS 的普通 GET 不受影响
+    with urllib.request.urlopen(f"http://127.0.0.1:{port}/tabs.html", timeout=5) as r:
+        assert r.status == 200 and b"tabbar" in r.read(), "普通静态页被 WS 路由挡住了"
+    print("PASS 1: 静态页照常服务")
+
+    # 2) 经代理的 motor/start 置位循环标志
+    post_json(port, "/ecat/api/motor/start")
+    assert flag() is True, "motor/start 经代理后没把 g_ecat_motor_live 置位"
+    print("PASS 2: 代理记账 motor/start → live=True")
+
+    # 3) 两条租约, 关一条 → 不该停
+    a = ws_connect(port)
+    b = ws_connect(port)
+    time.sleep(0.1)
+    a.close()
+    time.sleep(1.0)                     # > GRACE
+    assert "/api/op/stop" not in calls, "还剩一个窗口就停了循环!"
+    print("PASS 3: 仍有窗口在线时不停")
+
+    # 4) 全关 → GRACE 后自动 op/stop (走浏览器式 Close 帧回显路径,
+    #    与场景 3 的裸 FIN 路径各覆盖一种断开方式)
+    ws_graceful_close(b)
+    deadline = time.time() + 5
+    while time.time() < deadline and "/api/op/stop" not in calls:
+        time.sleep(0.05)
+    assert "/api/op/stop" in calls, "全部断开后没停循环"
+    assert wait_flag(False), "停完没清 live 标志"
+    print("PASS 4: 最后一条租约断开 → /api/op/stop")
+
+    # 5) 刷新吸收: 断开后在 GRACE 内重连 → 不该停
+    with calls_lock:
+        calls.clear()
+    post_json(port, "/ecat/api/motor/start")
+    c = ws_connect(port)
+    c.close()
+    d = ws_connect(port)                # 模拟刷新重连(远快于 GRACE)
+    time.sleep(1.0)
+    assert "/api/op/stop" not in calls, "页面刷新被误判成关闭 app!"
+    print("PASS 5: 刷新重连被 GRACE 吸收")
+
+    # 6) 用户显式 motor/stop 后, 关窗口不该再打 op/stop
+    d.close()
+    time.sleep(1.0)
+    with calls_lock:
+        calls.clear()
+    post_json(port, "/ecat/api/motor/start")
+    post_json(port, "/ecat/api/motor/stop")
+    assert flag() is False, "motor/stop 没清 live 标志"
+    e = ws_connect(port)
+    e.close()
+    time.sleep(1.0)
+    assert "/api/op/stop" not in calls, "循环本来就没跑还去停(多此一举)"
+    print("PASS 6: 循环已停时不再重复停")
+
+    # 7) 窗口模式竞态: 服务进程在 GRACE 定时器触发前就退出(/api/shutdown),
+    #    且租约连接还挂着 —— 退出路径必须同步把循环停掉, 不能指望定时器
+    #    (复用的 ecat_native 的 parent 不是本服务, 没人替它清场)。
+    srv.APP_LEASE_GRACE_S = 30.0      # 定时器"永不"到点, 证明停的是退出路径
+    srv.g_httpd = httpd
+    post_json(port, "/ecat/api/motor/start")
+    assert flag() is True
+    with calls_lock:
+        calls.clear()
+    f = ws_connect(port)              # 租约仍在(未断开)
+    post_json(port, "/api/shutdown")
+    deadline = time.time() + 5
+    while time.time() < deadline and "/api/op/stop" not in calls:
+        time.sleep(0.05)
+    assert "/api/op/stop" in calls, "服务进程退出前没停循环(窗口模式竞态)"
+    assert wait_flag(False, timeout=2.0)
+    f.close()
+    print("PASS 7: 服务进程退出时同步停循环")
+
+    httpd.shutdown()
+    print("ALL PASS")
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,175 @@
+/**
+ * test_app_lease_e2e.mjs — app 租约的真浏览器端到端验证。
+ *
+ * 与 test_app_lease.py 的分工: 那边用裸 TCP 验服务端逻辑; 这边用真 Edge
+ * (headless) 验 app-lease.js 在浏览器引擎里真的能握手/断连, 以及
+ * "多窗口只关一个不停、全关才停" 的真实时序。
+ *
+ * 不碰真硬件: 真 ecat_native 可能正占着 8090(它一收 motor/start 就会去动
+ * 真总线)。所以假后端放 8091, esp32_server 以"仅改 ECAT_PORTS/autostart
+ * 两个常量"的方式跑起来 —— Handler、代理、租约、GRACE 全是真实代码。
+ *
+ * 用法: node scripts/test_app_lease_e2e.mjs   (退出码 0 = 通过)
+ */
+import { spawn, execSync } from "node:child_process";
+import net from "node:net";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+const ROOT = path.resolve(import.meta.dirname, "..");
+const EDGE = "C:/Program Files (x86)/Microsoft/Edge/Application/msedge.exe";
+const FAKE_PORT = 8091;
+const DBG_PORT = 9334;
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "dcu-lease-"));
+const FAKE_LOG = path.join(tmp, "fake.log");
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+const procs = [];
+let failed = false;
+
+function check(name, cond, detail) {
+  if (cond) console.log("  ✓ " + name);
+  else { failed = true; console.log("  ✗ " + name + (detail ? " — " + detail : "")); }
+}
+function fakeLog() {
+  try { return fs.readFileSync(FAKE_LOG, "utf8"); } catch (_) { return ""; }
+}
+function waitPort(port, timeoutMs = 15000) {
+  return new Promise((res, rej) => {
+    const t0 = Date.now();
+    (function poll() {
+      const s = net.connect(port, "127.0.0.1");
+      s.on("connect", () => { s.destroy(); res(); });
+      s.on("error", () => {
+        if (Date.now() - t0 > timeoutMs) rej(new Error("port " + port + " never opened"));
+        else setTimeout(poll, 200);
+      });
+    })();
+  });
+}
+function freePort() {
+  return new Promise((res) => {
+    const s = net.createServer();
+    s.listen(0, "127.0.0.1", () => { const p = s.address().port; s.close(() => res(p)); });
+  });
+}
+function killTree(pid) {
+  try { execSync(`taskkill /PID ${pid} /T /F`, { stdio: "ignore" }); } catch (_) {}
+}
+
+/* ── 1. 假 EtherCAT 后端(8091): 记录请求路径, 一律回 ok:true ── */
+fs.writeFileSync(path.join(tmp, "fake_ecat.py"), `
+import sys
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+log = sys.argv[2]
+class H(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.0"
+    def do_POST(self):
+        n = int(self.headers.get("Content-Length") or 0)
+        self.rfile.read(n)
+        with open(log, "a") as f:
+            f.write(self.path + "\\n")
+        # 紧凑格式, 与真 native(C snprintf 打印)逐字节同构
+        body = b'{"ok":true,"log":["fake"]}'
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+    def log_message(self, *a):
+        pass
+ThreadingHTTPServer(("127.0.0.1", int(sys.argv[1])), H).serve_forever()
+`);
+procs.push(spawn("python", [path.join(tmp, "fake_ecat.py"), String(FAKE_PORT), FAKE_LOG], { stdio: "ignore" }));
+await waitPort(FAKE_PORT);
+
+/* ── 2. 真 esp32_server, 仅把代理端口表钉到 8091、禁掉 autostart ── */
+const srvPort = await freePort();
+fs.writeFileSync(path.join(tmp, "srv.py"), `
+import sys
+sys.path.insert(0, ${JSON.stringify(ROOT)})
+import esp32_server as srv
+srv.ECAT_PORTS = (${FAKE_PORT},)
+srv._ecat_autostart = lambda: (None, [])
+httpd, p = srv._bind_server(${srvPort})
+print("PORT=%d" % p, flush=True)
+httpd.serve_forever()
+`);
+const server = spawn("python", [path.join(tmp, "srv.py")], { cwd: ROOT });
+procs.push(server);
+let serverPort = 0;
+await new Promise((res, rej) => {
+  server.stdout.on("data", (d) => {
+    const m = /PORT=(\d+)/.exec(String(d));
+    if (m) { serverPort = Number(m[1]); res(); }
+  });
+  server.stderr.on("data", (d) => console.log("[srv] " + String(d).trim()));
+  setTimeout(() => rej(new Error("server 未报端口")), 15000);
+});
+const BASE = `http://127.0.0.1:${serverPort}`;
+
+/* ── 3. 真 Edge headless + CDP ── */
+const edge = spawn(EDGE, [
+  "--headless=new", `--remote-debugging-port=${DBG_PORT}`,
+  `--user-data-dir=${path.join(tmp, "profile")}`,
+  "--no-first-run", "--disable-gpu", "about:blank",
+], { stdio: "ignore" });
+procs.push(edge);
+await waitPort(DBG_PORT);
+
+let page;
+for (let i = 0; i < 40 && !page; i++) {
+  await sleep(250);
+  try {
+    const list = await (await fetch(`http://127.0.0.1:${DBG_PORT}/json/list`)).json();
+    page = list.find((t) => t.type === "page");
+  } catch (_) {}
+}
+if (!page) throw new Error("拿不到 Edge 调试页面目标");
+const cdp = new WebSocket(page.webSocketDebuggerUrl);
+await new Promise((r, j) => { cdp.onopen = r; cdp.onerror = j; });
+let mid = 0;
+const pend = new Map();
+cdp.onmessage = (e) => {
+  const m = JSON.parse(e.data);
+  if (m.id && pend.has(m.id)) { pend.get(m.id)(m); pend.delete(m.id); }
+};
+const send = (method, params = {}) => new Promise((res) => {
+  const id = ++mid; pend.set(id, res); cdp.send(JSON.stringify({ id, method, params }));
+});
+
+/* ── 4. 场景一: 单窗口, 关窗 → 自动停 ── */
+await send("Page.enable");
+await send("Page.navigate", { url: BASE + "/tabs.html" });
+await sleep(2500);
+const r1 = await fetch(BASE + "/ecat/api/motor/start", {
+  method: "POST", headers: { "content-type": "application/json" }, body: "{}" });
+check("motor/start 经代理成功", r1.ok && (await r1.json()).ok === true);
+check("假后端收到 /api/motor/start", fakeLog().includes("/api/motor/start"));
+await sleep(4500);   // > GRACE(3s): 窗口还开着, 绝不能停
+check("窗口在线期间没停循环", !fakeLog().includes("/api/op/stop"));
+await send("Page.navigate", { url: "about:blank" });  // 离开页面 → 租约断开
+await sleep(5000);
+check("关窗(GRACE 后)自动 /api/op/stop", fakeLog().includes("/api/op/stop"));
+
+/* ── 5. 场景二: 双窗口, 关一个不停, 全关才停 ── */
+fs.writeFileSync(FAKE_LOG, "");
+const t2 = await send("Target.createTarget", { url: BASE + "/tabs.html" });
+await sleep(2500);
+await fetch(BASE + "/ecat/api/motor/start", {
+  method: "POST", headers: { "content-type": "application/json" }, body: "{}" });
+await send("Page.navigate", { url: BASE + "/tabs.html" });  // 窗口1 回来续租
+await sleep(1500);
+await send("Target.closeTarget", { targetId: t2.result.targetId });
+await sleep(4500);
+check("还剩一个窗口时不停", !fakeLog().includes("/api/op/stop"));
+await send("Page.navigate", { url: "about:blank" });        // 最后一个也关掉
+await sleep(5000);
+check("全部窗口关闭后自动停", fakeLog().includes("/api/op/stop"));
+
+/* ── 6. 收尾 ── */
+try { cdp.close(); } catch (_) {}
+for (const p of procs) killTree(p.pid);
+try { fs.rmSync(tmp, { recursive: true, force: true }); } catch (_) {}
+console.log(failed ? "E2E FAIL" : "E2E ALL PASS");
+process.exit(failed ? 1 : 0);