From 13a1d971bedd6e5af4697afb4abfbc7cccbfecfa Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Mon, 21 Sep 2026 18:28:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=BF=9E=E6=8E=A5=E5=AF=BF=E5=91=BD?= =?UTF-8?q?=E5=88=B0=E7=82=B9=E5=89=8D=E4=B8=BB=E5=8A=A8=E9=87=8D=E8=BF=9E?= =?UTF-8?q?=EF=BC=88=E6=9C=8D=E5=8A=A1=E7=AB=AF=E7=BA=A6=201=20=E5=B0=8F?= =?UTF-8?q?=E6=97=B6=E4=BC=9A=E6=97=A0=E9=A2=84=E8=AD=A6=E6=96=AD=E5=BC=80?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实测 QQ 服务端在一个连接建立约 1 小时后会自己断开。此前只能等它断： 要么被看门狗判成静默超时、要么等读写失败，都算"异常断开"，观感与恢复 路径都不理想。现改为临到期前主动收尾重连——是正常断开 + RESUME 续传， 期间消息由网关补发。 实现：receive_helper 记录寿命截止点（_SERVER_LIFETIME - _RECONNECT_MARGIN， 默认 60min - 10min = 50min），收帧超时取"静默阈值"与"寿命余量"两个截止里 的先到者——静默过久仍抛 TimeoutError 交给看门狗，寿命到点则 return 正常 结束本轮连接，run() 见存活超过 _STABLE_SECONDS 便立刻重连（沿用 b37a53f 的"健康断开零等待"路径），RESUME 带上 session_id 续传。 _connect 的存活秒数天然 >= 50min > _STABLE_SECONDS，主动重连必被判定为 健康断开，无需额外标志位。 测试：新增 test_proactive_reconnect_before_server_lifetime（真实 ws 端到端， 把寿命阈值调到 0.5s 让这一刻立即发生）；已验证把寿命阈值关掉后该测试会 超时失败，不是空转到通过。另给 tests/test_reconnect.py 补 CHANGELOG 条目 （b37a53f 漏记），并同步 connecter 模块 docstring / ARCHITECTURE / GUIDE。 90 测试全绿，ruff check/format + pyright 干净。 --- CHANGELOG.md | 1 + docs/ARCHITECTURE.md | 2 +- docs/GUIDE.md | 2 +- src/qqbotsdk/connecter.py | 31 +++++++++++--- tests/test_reconnect.py | 4 +- tests/test_ws_integration.py | 83 ++++++++++++++++++++++++++++++------ 6 files changed, 100 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 32d8e2b..547df7b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ - handler 后台并发：`@ee.on(..., background=True)` 把 handler 放进后台任务，不阻塞事件流（同事件内失去先后保证，返回值不回流）；`run_loop` 退出经 `emitter.close()` 统一取消在飞任务 - OpenAPI 限流与鉴权自愈：`BotApi.request` 对 429 自动重试（最多 3 次，优先 `Retry-After`，脏值容错解析，退避封顶 30s）；401 强制刷新 token 后重试一次，token 被服务端提前作废不再直接抛错 - 群消息全量模式：新增 `GROUP_MESSAGE_CREATE` 事件接收，与 `GROUP_AT_MESSAGE_CREATE` 共用 `GroupAtMessage`（需在开放平台开通"接收所有消息"权限，不限 @机器人）；`GroupAtMessage` 补齐官方新增的 `message_type`/`message_scene`/`attachments`/`mentions`/`ark_data`/`msg_elements` 字段 +- **连接寿命主动重连**：服务端约 1 小时后会无预警断开连接（实测），故连接存活到 `_SERVER_LIFETIME - _RECONNECT_MARGIN`（默认 60min − 10min = 50min）时由 SDK 主动收尾重连——走正常 RESUME 续传、期间消息由网关补发，而不是被服务端掐断后由看门狗判成异常断开；收帧超时在"静默阈值"与"寿命余量"两个截止里取先到者，静默仍归看门狗处理 - 文档站点：基于 docsify（侧边栏、全文搜索、代码复制、mermaid 图渲染、分页），`npx docsify-cli serve` 本地预览；新增 `examples/README.md` 示例索引 ### 变更 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 37699c0..9739ca4 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -108,7 +108,7 @@ sequenceDiagram 1. **业务事件走管线，协议帧走专有处理器**：只有 op=0 的业务事件经 `EventQueue` 异步流转给 `EventEmitter` 分发；HELLO/READY/INVALID_SESSION/op=13 等协议帧由 connecter 逐帧调用其专有协议处理器（`BaseProtocol.on_frame`）就地处理。这样用户 handler 不会看到协议帧，且"断线前重置会话"这类时序不再取决于分发泵的进度——`INVALID_SESSION` 的会话清空在 connecter `break` 重连前已同步完成，不会误 RESUME 一个已死会话。代价是多一跳队列延迟（可忽略）。 2. **协议层只管会话语义，心跳归传输层**：`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号（说什么），`WebsocketConnecter` 负责连接与心跳（怎么保活）——心跳任务随连接生灭（HELLO 给出间隔后启动，断线即取消），断线期间不在出站缓冲堆积过期心跳，也免去了调度器的生命周期管理。协议处理器的应答经 `on_frame` 返回值交给 connecter：ws 与心跳一起进 connecter 私有的出站缓冲、由单任务发送（维持 ws 单写者不变量），webhook 则直接写进 HTTP 响应体。 -3. **看门狗两层互补**（见 `connecter.py` 模块注释）：静默超时抓连接彻底失联，ACK 期限检查抓"有事件流但心跳已死"。阈值 `_SILENCE_FACTOR=1.1`、`_ACK_FACTOR=1.2` 是按 ACK 自然节奏校准的，**不能压到 1× 整**，否则正常抖动会周期性误杀（误杀由 RESUME 兜底，不丢消息）。重连退避只在连接稳定存活（`_STABLE_SECONDS`）后才归一，抖动的服务端不会被 1s 间隔反复探测。 +3. **看门狗两层互补 + 连接寿命主动重连**（见 `connecter.py` 模块注释）：静默超时抓连接彻底失联，ACK 期限检查抓"有事件流但心跳已死"。阈值 `_SILENCE_FACTOR=1.1`、`_ACK_FACTOR=1.2` 是按 ACK 自然节奏校准的，**不能压到 1× 整**，否则正常抖动会周期性误杀（误杀由 RESUME 兜底，不丢消息）。重连退避只在连接稳定存活（`_STABLE_SECONDS`）后才归一，抖动的服务端不会被 1s 间隔反复探测。此外服务端约 1 小时会**无预警**断开连接（实测），故连接存活到 `_SERVER_LIFETIME - _RECONNECT_MARGIN`（默认 50 分钟）时 SDK 主动收尾重连——收帧超时取"静默阈值"与"寿命余量"两个截止中的先到者，静默仍交给看门狗，寿命到点则正常结束本轮连接、由 `run()` 判定为健康断开后立刻 RESUME 续传。 4. **handler 异常就地隔离**：`_invoke` 捕获记日志，`dispatch` 主循环另有兜底；单条事件的单个 handler 失败不影响进程。`background=True` 的 handler 进后台任务（emit 不等它），任务由 emitter 持强引用、`close()` 统一取消。 5. **注入约定**：handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象；`Event` → 事件封装；登记进 `emitter.services` 的组件类型（如 `BotApi`）→ 实例；未标注或未登记 → 回退传**原始 d**。解析是纯同步查表，无框架、无运行时魔法：每个 handler 的 `get_type_hints` + 形参名只解析一次并缓存，但 `services` 命中在每条事件时实时判定，保证"先注册 handler、后登记组件"的装配顺序可用。 6. **事件注册不依赖第三方库**：`EventEmitter` 内部就是一个 `事件名 → handler 列表` 的字典，`on()` 追加、`emit()` 顺序遍历。不需要 pyee 那套 `once`/`new_listener`/错误事件机制（SDK 用不到），少一个依赖、语义也一眼可读。注册语义：同一函数对同一事件重复注册只生效一次（沿用原先 pyee 的行为），分发按注册顺序串行执行（`background=True` 的除外）。 diff --git a/docs/GUIDE.md b/docs/GUIDE.md index f6f6272..b974e41 100644 --- a/docs/GUIDE.md +++ b/docs/GUIDE.md @@ -74,7 +74,7 @@ QQ 网关只认分组位掩码，**组内单个事件无法单独开关**—— |---|---|---| | 网络要求 | 能访问外网即可，本地开发友好 | 需公网可达 + HTTPS，仅支持 80/443/8080/8443 端口 | | 配置 | 无额外配置 | 平台回调地址填 `https://<域名>:<端口>` | -| 连接维护 | SDK 自动心跳、断线指数退避重连并 RESUME 续传 | SDK 处理 Ed25519 验签、op=13 验证应答、60s TTL 事件去重 | +| 连接维护 | SDK 自动心跳、断线指数退避重连并 RESUME 续传；服务端约 1 小时会断开连接，SDK 会在临到期前（约 50 分钟）主动重连续传 | SDK 处理 Ed25519 验签、op=13 验证应答、60s TTL 事件去重 | | 适用 | 长驻进程、需要会话与重连保障 | 无状态部署、回调型服务 | 两种方式下 handler 的写法完全一致，切换只需改 `CONNECTER`。 diff --git a/src/qqbotsdk/connecter.py b/src/qqbotsdk/connecter.py index 5f9802f..a3aac89 100644 --- a/src/qqbotsdk/connecter.py +++ b/src/qqbotsdk/connecter.py @@ -1,4 +1,4 @@ -"""连接适配器：把两种接入方式（WebSocket / Webhook）归一为事件流入、reply 队列流出。 +"""连接适配器：把两种接入方式（WebSocket / Webhook）归一为业务事件流入、出站帧流出。 每条帧先交给本适配器注入的协议处理器（`BaseProtocol.on_frame`）处理协议语义， 只有业务事件（op=0）才入队交给 emitter 分发给用户 handler——协议帧与业务事件 @@ -6,12 +6,13 @@ - WebsocketConnecter：gateway 握手 + ws 长连接，指数退避重连（连接稳定 存活才把退避归一，抖动的服务端不会被 1s 间隔反复探测）；心跳任务随 - 连接生灭——HELLO 给出间隔后周期投递，断线即取消，不在 reply 队列堆积 + 连接生灭——HELLO 给出间隔后周期投递，断线即取消，不在出站缓冲堆积 过期心跳；看门狗两层互补——receive 的静默超时抓"连接彻底静默"， - 收到消息时的 ACK 期限检查抓"有事件流但心跳已死"； + 收到消息时的 ACK 期限检查抓"有事件流但心跳已死"；另外服务端约 1 小时 + 会主动断开连接，故临到期前主动重连（经 RESUME 补发消息），不等被掐断； - WebhookConnecter：aiohttp server 接收回调，验签失败 401， 按 payload id 做 60s TTL 去重，正常推送进事件队列；op=13 验证请求 - 由协议处理器就地应答（HTTP 响应不走 reply 队列）。 + 由协议处理器就地应答（HTTP 响应不走队列）。 """ import asyncio @@ -51,6 +52,10 @@ class WebsocketConnecter(Connecter): _ACK_FACTOR = 1.2 # 连接存活超过该时长才算稳定，才把退避归一 _STABLE_SECONDS = 60.0 + # 服务端约 1 小时后会主动断开连接（无预警，实测）：临到期前留出余量主动重连， + # 这样是正常收尾 + RESUME 补发消息，而不是被掐断后由看门狗判成异常 + _SERVER_LIFETIME = 60 * 60.0 + _RECONNECT_MARGIN = 10 * 60.0 def __init__( self, @@ -147,9 +152,23 @@ class WebsocketConnecter(Connecter): ) -> None: interval: float = 0 last_ack = monotonic() + # 服务端约 1 小时后会主动断开，临到期前主动重连：return 即让 _connect 正常 + # 收尾，run() 见存活超过稳定阈值便立刻重连，经 RESUME 补发这段时间的消息 + deadline = monotonic() + self._SERVER_LIFETIME - self._RECONNECT_MARGIN while not ws.closed: - # HELLO 前不设超时；之后按看门狗阈值等待任何消息 - msg = await ws.receive(timeout=interval * self._SILENCE_FACTOR or None) + remaining = deadline - monotonic() + if remaining <= 0: + logger.info("连接到达服务端寿命上限，主动重连") + return + # 两个截止取先到者：静默阈值盯"连接静默"，寿命余量保证临到期准时醒来 + silence = interval * self._SILENCE_FACTOR + watchdog = bool(silence) and silence < remaining + try: + msg = await ws.receive(timeout=silence if watchdog else remaining) + except TimeoutError: + if watchdog: + raise # 静默过久 → 交给 run() 走看门狗重连 + continue # 寿命余量到点 → 回顶部主动收尾 if msg.type != WSMsgType.TEXT: logger.warning(f"WebSocket 收到非文本帧，连接终止: {msg.type}") break diff --git a/tests/test_reconnect.py b/tests/test_reconnect.py index fdf8cf2..a48339a 100644 --- a/tests/test_reconnect.py +++ b/tests/test_reconnect.py @@ -64,9 +64,7 @@ async def test_failed_reconnect_backs_off_and_logs_disconnect( monkeypatch: pytest.MonkeyPatch, ) -> None: """连不上时记断开并按退避等待，退避逐次翻倍。""" - sleeps, logs = await _run_capture( - monkeypatch, [3600.0, TimeoutError(), 0.0, 0.0] - ) + sleeps, logs = await _run_capture(monkeypatch, [3600.0, TimeoutError(), 0.0, 0.0]) # 健康断开 → 无 sleep；失败后依次 1s、2s、4s assert sleeps == [1, 2, 4] diff --git a/tests/test_ws_integration.py b/tests/test_ws_integration.py index d92c228..aa24daa 100644 --- a/tests/test_ws_integration.py +++ b/tests/test_ws_integration.py @@ -25,11 +25,15 @@ class FakeGateway: def __init__(self) -> None: self.identifies: list[dict] = [] self.heartbeats = 0 + self.resumes: list[dict] = [] + self.connections = 0 self.runner: web.AppRunner | None = None self.port = 0 # 测试用同步点 self.ready_sent = asyncio.Event() self.got_dispatch = asyncio.Event() + self.second_connection = asyncio.Event() + self.resume_received = asyncio.Event() async def start(self) -> None: app = web.Application() @@ -56,6 +60,9 @@ class FakeGateway: async def _ws(self, request: web.Request) -> web.WebSocketResponse: ws = web.WebSocketResponse() await ws.prepare(request) + self.connections += 1 + if self.connections >= 2: + self.second_connection.set() await ws.send_json({"op": 10, "d": {"heartbeat_interval": 300}}) s = 0 @@ -87,29 +94,49 @@ class FakeGateway: "d": {"content": "ping", "group_openid": "G1"}, } ) + elif op == 6: # RESUME + self.resumes.append(payload) + self.resume_received.set() elif op == 1: # HEARTBEAT self.heartbeats += 1 await ws.send_json({"op": 11}) return ws +def make_config(gw: FakeGateway) -> Config: + return Config( + app_id="app", + app_secret="secret", + base_url=f"http://127.0.0.1:{gw.port}", + timeout=None, # type: ignore[arg-type] + connecter="websocket", + intents_file="no-such-intents.toml", # 测试环境无配置，intents=0 + webhook_host="127.0.0.1", + webhook_port=0, + webhook_path="/x", + ) + + +def build_connecter( + config: Config, +) -> tuple[ClientSession, EventEmitter, WebsocketConnecter]: + http = ClientSession(timeout=config.timeout) + token = AccessToken(config, http) + session = Session() + emitter = EventEmitter(EventQueue()) + protocol = WebsocketProtocol(config, session, token) + connecter = WebsocketConnecter( + config, http, token, session, emitter.queue, protocol + ) + return http, emitter, connecter + + @pytest.mark.asyncio async def test_websocket_end_to_end(): gw = FakeGateway() await gw.start() try: - config = Config( - app_id="app", - app_secret="secret", - base_url=f"http://127.0.0.1:{gw.port}", - timeout=None, # type: ignore[arg-type] - connecter="websocket", - intents_file="no-such-intents.toml", # 测试环境无配置，intents=0 - webhook_host="127.0.0.1", - webhook_port=0, - webhook_path="/x", - ) - + config = make_config(gw) http = ClientSession(timeout=config.timeout) token = AccessToken(config, http) queue = EventQueue() @@ -165,3 +192,35 @@ async def test_websocket_end_to_end(): await http.close() finally: await gw.stop() + + +@pytest.mark.asyncio +async def test_proactive_reconnect_before_server_lifetime(): + """服务端约 1 小时后会无预警掐断连接：应在临到期前主动重连。 + + 把寿命阈值压到极短让这一刻在测试里立刻发生。第二次连接必须是 RESUME + ——说明是正常收尾后按会话续传（服务端补发期间消息），而不是被掐断后 + 从头 IDENTIFY。 + """ + gw = FakeGateway() + await gw.start() + try: + config = make_config(gw) + http, _emitter, connecter = build_connecter(config) + # 0.5s 寿命 - 0.3s 余量 = 连接存活约 0.2s 后主动重连；"稳定"阈值一并调小， + # 让这次主动重连被判定为正常收尾（立即重连，不退避） + connecter._SERVER_LIFETIME = 0.5 + connecter._RECONNECT_MARGIN = 0.3 + connecter._STABLE_SECONDS = 0.1 + connect_task = asyncio.create_task(connecter.run()) + try: + await asyncio.wait_for(gw.resume_received.wait(), timeout=10) + assert gw.connections >= 2 # 确实重连过 + assert len(gw.identifies) == 1 # 只 IDENTIFY 过一次 + assert gw.resumes[0]["d"]["session_id"] == "S1" # 续传同一会话 + finally: + connect_task.cancel() + await asyncio.gather(connect_task, return_exceptions=True) + await http.close() + finally: + await gw.stop()