zijunchiang/opendance-windows
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

feat(net): 连接保活 / 设备去重 / 连接状态机（T-WIN-015，契约 websocket-keypoints 0.2.0）

Browse Source 
与 Android 端 T-AND-011「两端一起改」：契约先在 opendance-web 修改并提升到
websocket-keypoints 0.2.0（向后兼容：子协议名与二进制帧 version 均不变，
只新增可选控制字段与一种新控制类型），再同步三端快照。

修掉本端的 3 处根因/放大器：
1. 静默超时 5s 过短且未声明，客户端无从得知。OkHttp 的协议层 ping 由底层库
   自动应答、**不暴露给应用层**，无法刷新 last_seen_ms，于是"画面里没人"时
   每 5 秒必然踢一次连接 -> 客户端重连 -> 再被踢。
   - CLIENT_TIMEOUT_MS 5s -> 15s，新增 HEARTBEAT_INTERVAL_MS = 5s；
   - welcome_payload() 在 welcome 中声明 silentTimeoutMs / heartbeatIntervalMs，
     config 广播同样带上（契约 §1.1：心跳必须 < 静默超时的一半，5000*2 < 15000）；
   - 处理应用层 ping 并回 pong（回显 sentAtMs + serverAtMs，供对端算 RTT）。
2. 反复重连被计成多台设备。按契约 §5 用 hello.deviceId 去重：同一设备只保留
   最新连接，旧连接以 close(1000) 关闭——刻意用 1000，因为对端按契约 §7
   收到 1000 不会重连，否则会形成"挤掉-重连-再挤掉"的循环；
   新增 device_count() 与 device_replaced 信号，界面一律显示去重后的设备数。
3. 只有 handshaken 布尔量，日志看不出卡在哪一步。按契约 §8 逐客户端维护与
   Android 端同名的状态（connecting -> handshaking -> streaming -> degraded，
   主动停止/超时/被替换为 stopped），并记录 close code/reason、协商子协议、
   连接/握手/首帧时间；degraded 保持时长做成可覆盖字段便于测试。

验收测试（tests/test_pose_server.gd 新增 7 项，总计 720 通过 / 0 失败）：
welcome 声明保活参数、ping->pong 回显、同 deviceId 去重（旧连接 1000 关闭且
设备数仍为 1）、8 个状态名与契约一致、状态随握手/首帧推进到 streaming、
丢帧进入 degraded 后恢复、snapshot 暴露设备数与保活参数。

同时更新 README/README.txt/tasks 与 Credits 的契约版本（0.2.0）。
This commit is contained in:
zijunchiang
2026-09-17 01:11:48 +08:00
parent 5581cbe05d
commit c3a227c6df
12 changed files with 647 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_pose_server.gd
+189
View File
@@ -35,6 +35,8 @@ func _setup() -> void:
_server = PoseServer.new()
_server.subprotocol = SUBPROTOCOL
# 缩短 degraded 保持时长：无头测试不能依赖真实等待 3 秒
_server.degraded_hold_ms = 60
_tree().root.add_child(_server)
_server.frame_received.connect(func(frame): _received.append(frame))
_server.control_received.connect(func(_id, msg): _controls.append(msg))
@@ -293,3 +295,190 @@ func test_stop_closes_listener() -> void:
is_false(_server.is_running(), "stop() 后不再监听")
eq(_server.connected_count(), 0, "stop() 后连接清空")
await _teardown()
# ---------------- T-AND-011：契约 v0.2.0 的保活 / 去重 / 状态机 ----------------
## 连入第二个客户端（用于 deviceId 去重测试）
func _open_second_client() -> WebSocketPeer:
var peer := WebSocketPeer.new()
peer.supported_protocols = PackedStringArray([SUBPROTOCOL])
peer.connect_to_url("ws://127.0.0.1:%d/pose" % TEST_PORT)
var guard := 0
while peer.get_ready_state() == WebSocketPeer.STATE_CONNECTING and guard < 240:
peer.poll()
await _pump()
guard += 1
return peer
func _send_hello(peer: WebSocketPeer, device_id: String, device: String = "测试机") -> void:
peer.send_text(JSON.stringify({
"type": "hello",
"protocol": SUBPROTOCOL,
"deviceId": device_id,
"device": device,
"appVersion": "0.2.0",
"cameraFacing": "front",
"targetFps": 30,
"resolution": "640x480",
}))
## 等某个客户端收到指定类型的控制帧
func _await_control(peer: WebSocketPeer, type: String, max_frames: int = 90) -> Dictionary:
for i in max_frames:
peer.poll()
await _pump()
while peer.get_available_packet_count() > 0:
var packet := peer.get_packet()
if not peer.was_string_packet():
continue
var parsed: Variant = JSON.parse_string(packet.get_string_from_utf8())
if typeof(parsed) == TYPE_DICTIONARY and str((parsed as Dictionary).get("type", "")) == type:
return parsed
return {}
func test_welcome_declares_keepalive_parameters() -> void:
await _setup()
await _pump(3)
_send_hello(_client, "device-keepalive")
var welcome := await _await_control(_client, "welcome")
is_false(welcome.is_empty(), "收到 welcome 控制帧")
var timeout_ms := int(welcome.get("silentTimeoutMs", 0))
var heartbeat_ms := int(welcome.get("heartbeatIntervalMs", 0))
ok(timeout_ms > 0, "welcome 声明 silentTimeoutMs", str(timeout_ms))
ok(heartbeat_ms > 0, "welcome 声明 heartbeatIntervalMs", str(heartbeat_ms))
# 契约 §1.1 硬性要求：心跳间隔必须小于静默超时的一半
ok(heartbeat_ms * 2 < timeout_ms, "心跳间隔小于静默超时的一半",
"心跳 %d ms，超时 %d ms" % [heartbeat_ms, timeout_ms])
await _teardown()
func test_ping_gets_pong_with_echoed_timestamp() -> void:
await _setup()
await _pump(3)
_send_hello(_client, "device-ping")
await _await_control(_client, "welcome")
_client.send_text(JSON.stringify({"type": "ping", "sentAtMs": 1730000000123}))
var pong := await _await_control(_client, "pong")
is_false(pong.is_empty(), "收到 pong 控制帧（应用层心跳可续命）")
eq(int(pong.get("sentAtMs", 0)), 1730000000123, "pong 原样回传 sentAtMs")
ok(int(pong.get("serverAtMs", 0)) > 0, "pong 含服务端时间戳")
await _teardown()
func test_duplicate_device_id_is_deduplicated() -> void:
await _setup()
await _pump(3)
_send_hello(_client, "device-dup", "第一台")
await _await_control(_client, "welcome")
await _pump(3)
eq(_server.device_count(), 1, "去重前设备数为 1")
var second := await _open_second_client()
eq(second.get_ready_state(), WebSocketPeer.STATE_OPEN, "第二台客户端握手成功")
_send_hello(second, "device-dup", "第一台（重连）")
await _await_control(second, "welcome")
# 服务端应在若干帧内关掉旧连接，并以 1000 关闭（对端不重连）
var guard := 0
while guard < 120 and _client.get_ready_state() != WebSocketPeer.STATE_CLOSED:
_client.poll()
await _pump()
guard += 1
eq(_client.get_ready_state(), WebSocketPeer.STATE_CLOSED, "同一 deviceId 的旧连接被关闭")
eq(_client.get_close_code(), 1000, "旧连接以 1000 关闭（契约 §7：对端不重连）")
eq(_server.device_count(), 1, "去重后设备数仍为 1（不因重连变多）")
is_true(_server.connected_count() >= 1, "新连接仍在")
second.close()
await _pump(2)
await _teardown()
func test_states_follow_contract_names() -> void:
eq(PoseServer.state_name(PoseServer.ClientState.IDLE), "idle", "idle 状态名")
eq(PoseServer.state_name(PoseServer.ClientState.RESOLVING), "resolving", "resolving 状态名")
eq(PoseServer.state_name(PoseServer.ClientState.CONNECTING), "connecting", "connecting 状态名")
eq(PoseServer.state_name(PoseServer.ClientState.HANDSHAKING), "handshaking", "handshaking 状态名")
eq(PoseServer.state_name(PoseServer.ClientState.STREAMING), "streaming", "streaming 状态名")
eq(PoseServer.state_name(PoseServer.ClientState.DEGRADED), "degraded", "degraded 状态名")
eq(PoseServer.state_name(PoseServer.ClientState.RECONNECTING), "reconnecting", "reconnecting 状态名")
eq(PoseServer.state_name(PoseServer.ClientState.STOPPED), "stopped", "stopped 状态名")
func test_client_state_progresses_to_streaming() -> void:
await _setup()
await _pump(3)
_send_hello(_client, "device-state")
await _await_control(_client, "welcome")
var info_before: Array = _server.clients_info()
eq(info_before.size(), 1, "clients_info 含 1 个客户端")
eq(str(info_before[0]["state"]), "handshaking", "只发 hello 时为 handshaking")
_client.send(KeypointFrame.encode(0, _sample_points(), _sample_confidence()))
var guard := 0
while _received.is_empty() and guard < 90:
_client.poll()
await _pump()
guard += 1
var info: Array = _server.clients_info()
eq(str(info[0]["state"]), "streaming", "收到首帧后进入 streaming")
eq(str(info[0]["device_id"]), "device-state", "deviceId 被记录")
eq(str(info[0]["protocol"]), SUBPROTOCOL, "记录协商后的子协议")
ok(int(info[0]["first_frame_at_ms"]) > 0, "记录首帧时间（可诊断）")
await _teardown()
func test_frame_gap_marks_degraded_then_recovers() -> void:
await _setup()
await _pump(3)
_send_hello(_client, "device-degraded")
await _await_control(_client, "welcome")
_client.send(KeypointFrame.encode(0, _sample_points(), _sample_confidence()))
_client.send(KeypointFrame.encode(9, _sample_points(0.01), _sample_confidence()))
var guard := 0
while _server.frames_dropped == 0 and guard < 120:
_client.poll()
await _pump()
guard += 1
eq(_server.frames_dropped, 8, "跳号 1~8 被计为丢包")
eq(str(_server.clients_info()[0]["state"]), "degraded", "出现丢帧时进入 degraded")
# 无新丢帧持续 DEGRADED_HOLD_MS 后回到 streaming
var waited := 0
while waited < 400 and str(_server.clients_info()[0]["state"]) != "streaming":
_client.poll()
await _pump()
waited += 1
eq(str(_server.clients_info()[0]["state"]), "streaming", "丢帧停止后回到 streaming")
await _teardown()
func test_snapshot_exposes_keepalive_and_device_count() -> void:
await _setup()
await _pump(3)
_send_hello(_client, "device-snapshot")
await _await_control(_client, "welcome")
await _pump(3)
var snap := _server.snapshot()
eq(int(snap["devices"]), 1, "snapshot 报告去重后设备数")
eq(int(snap["silent_timeout_ms"]), PoseServer.CLIENT_TIMEOUT_MS, "snapshot 报告静默超时")
eq(int(snap["heartbeat_interval_ms"]), PoseServer.HEARTBEAT_INTERVAL_MS, "snapshot 报告心跳间隔")
await _teardown()