From 0c640be189e0b3267d4ae832a994d63b8e17bcd6 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 13 Oct 2026 07:13:01 +0800 Subject: [PATCH] fix(qqbot): reconnect when the gateway requests a new connection --- app/modules/qqbot/gateway.py | 11 ++- tests/test_qq_gateway_reconnect.py | 108 +++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+), 3 deletions(-) create mode 100644 tests/test_qq_gateway_reconnect.py diff --git a/app/modules/qqbot/gateway.py b/app/modules/qqbot/gateway.py index faa61cf2b..3e8e1232c 100644 --- a/app/modules/qqbot/gateway.py +++ b/app/modules/qqbot/gateway.py @@ -39,6 +39,7 @@ def run_gateway( :param ws_holder: 调用方持有的单元素列表，存放当前 WebSocketApp，供 stop() 时 close 以打断 run_forever """ last_seq: Optional[int] = None + attempt = 0 heartbeat_thread: Optional[threading.Thread] = None heartbeat_stop_event: Optional[threading.Event] = None heartbeat_epoch = 0 @@ -148,7 +149,7 @@ def run_gateway( def on_ws_message(ws, message): """处理当前 WebSocket 连接收到的 QQ Gateway 消息。""" - nonlocal last_seq + nonlocal last_seq, attempt try: payload = json.loads(message) except json.JSONDecodeError as err: @@ -186,9 +187,11 @@ def run_gateway( elif op == 0: # Dispatch if t == "READY": + attempt = 0 session_id = d.get("session_id", "") logger.info(f"[QQ Gateway:{config_name}] 连接成功 Ready, session_id={session_id}") elif t == "RESUMED": + attempt = 0 logger.info(f"[QQ Gateway:{config_name}] 连接成功 Session resumed") elif t == "C2C_MESSAGE_CREATE": author = d.get("author", {}) @@ -228,7 +231,8 @@ def run_gateway( elif op == 7: # Reconnect logger.info(f"[QQ Gateway:{config_name}] Reconnect requested") - # 当前实现不自动重连，由外层循环处理 + # run_forever 只有退出后外层才会重连，不能仅记录服务端指令。 + ws.close() elif op == 9: # Invalid Session logger.warning(f"[QQ Gateway:{config_name}] Invalid session") @@ -250,7 +254,6 @@ def run_gateway( stop_heartbeat(wait=False) reconnect_delays = [1, 2, 5, 10, 30, 60] - attempt = 0 try: while not stop_event.is_set(): @@ -266,6 +269,8 @@ def run_gateway( on_close=on_ws_close, ) with heartbeat_state_lock: + # 当前每次都重新 Identify，不能携带上一会话的心跳序号。 + last_seq = None ws_holder.clear() ws_holder.append(ws) diff --git a/tests/test_qq_gateway_reconnect.py b/tests/test_qq_gateway_reconnect.py new file mode 100644 index 000000000..9ec5663a5 --- /dev/null +++ b/tests/test_qq_gateway_reconnect.py @@ -0,0 +1,108 @@ +"""验证 QQ Gateway 重连与 HTTP 主动通知的独立生命周期。""" + +import json +import threading +from unittest.mock import Mock + +import pytest + +from app.modules.qqbot import gateway, qqbot + + +@pytest.mark.parametrize("opcode", [7, 9]) +@pytest.mark.parametrize("ready_event", ["READY", "RESUMED"]) +def test_gateway_reconnect_restores_dispatch_and_resets_backoff( + monkeypatch: pytest.MonkeyPatch, opcode: int, ready_event: str, +) -> None: + """服务端重连指令必须退出旧连接，成功鉴权后恢复短退避及入站派发。""" + stop_event = threading.Event() + sockets = [] + sleeps = [] + received = Mock() + token = Mock(return_value="token") + heartbeat_sequences = [] + + def build_websocket(_url, **callbacks): + """用四轮连接覆盖成功、鉴权前失败和再次成功后的重连。""" + ws = Mock() + sockets.append(ws) + index = len(sockets) + heartbeat_received = threading.Event() + + def send(message): + """记录新连接的首个心跳，检查旧会话序号不会泄漏。""" + payload = json.loads(message) + if payload["op"] == 1: + if not heartbeat_received.is_set(): + heartbeat_sequences.append(payload["d"]) + heartbeat_received.set() + + ws.send.side_effect = send + + def run_forever(**_kwargs): + """模拟完整协议回调，不连接真实 QQ 服务。""" + callbacks["on_message"](ws, json.dumps({ + "op": 10, "d": {"heartbeat_interval": 1}, + })) + assert heartbeat_received.wait(1) + if index != 2: + callbacks["on_message"](ws, json.dumps({ + "op": 0, "t": ready_event, "s": 42, + "d": {"session_id": "session"}, + })) + if index < 4: + callbacks["on_message"](ws, json.dumps({"op": opcode})) + ws.close.assert_called_once() + callbacks["on_close"](ws, 1000, "reconnect") + else: + callbacks["on_message"](ws, json.dumps({ + "op": 0, "t": "C2C_MESSAGE_CREATE", "s": 43, + "d": {"id": "message", "content": "hello", + "author": {"user_openid": "user"}}, + })) + stop_event.set() + + ws.run_forever.side_effect = run_forever + return ws + + monkeypatch.setattr(gateway.websocket, "WebSocketApp", build_websocket) + monkeypatch.setattr(gateway.time, "sleep", sleeps.append) + gateway.run_gateway( + "app", "secret", "test", token, + lambda _token: "wss://gateway.test", received, stop_event, [], + ) + + assert len(sockets) == token.call_count == 4 + # 第一轮成功后 1 秒，鉴权前失败后 2 秒，第三轮成功后恢复为 1 秒。 + assert sleeps == [0.1] * 40 + assert heartbeat_sequences == [None] * 4 + received.assert_called_once_with({ + "type": "C2C_MESSAGE_CREATE", "content": "hello", + "author": {"user_openid": "user"}, "id": "message", "timestamp": "", + }) + for ws in sockets: + assert json.loads(ws.send.call_args_list[0].args[0])["op"] == 2 + + +@pytest.mark.parametrize("is_group", [False, True]) +def test_proactive_notification_does_not_depend_on_gateway( + monkeypatch: pytest.MonkeyPatch, is_group: bool, +) -> None: + """Gateway 已断开时仍通过 HTTP 向配置的用户或群发送主动通知。""" + client = qqbot.QQBot.__new__(qqbot.QQBot) + client._ready = True + client._app_id = "app" + client._app_secret = "secret" + client._default_openid = None if is_group else "user" + client._default_group_openid = "group" if is_group else None + client._gateway_ws_holder = [] + send = Mock(return_value={"id": "sent"}) + monkeypatch.setattr(qqbot, "get_access_token", Mock(return_value="token")) + monkeypatch.setattr( + qqbot, + "send_proactive_group_message" if is_group else "send_proactive_c2c_message", + send, + ) + + assert client.send_msg(title="入库通知", text="整理完成") is True + assert send.call_args.args[:2] == ("token", "group" if is_group else "user")