mirror of
https://github.com/snowy-yang/qqbotsdk.git
synced 2026-10-21 12:24:49 +00:00
feat: 连接寿命到点前主动重连（服务端约 1 小时会无预警断开）
实测 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 干净。
This commit is contained in:
snowy-yang committed
1 parent
b37a53f6e0
commit
13a1d971be
6 files changed
+100
-23
No files matched your search
@@ -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` 示例索引
|
||||
|
||||
### 变更
|
||||
|
||||
@@ -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` 的除外）。
|
||||
|
||||
+1
-1
@@ -74,7 +74,7 @@ QQ 网关只认分组位掩码，**组内单个事件无法单独开关**——
|
||||
|---|---|---|
|
||||
| 网络要求 | 能访问外网即可，本地开发友好 | 需公网可达 + HTTPS，仅支持 80/443/8080/8443 端口 |
|
||||
| 配置 | 无额外配置 | 平台回调地址填 `https://<域名>:<端口><WEBHOOK_PATH>` |
|
||||
| 连接维护 | SDK 自动心跳、断线指数退避重连并 RESUME 续传 | SDK 处理 Ed25519 验签、op=13 验证应答、60s TTL 事件去重 |
|
||||
| 连接维护 | SDK 自动心跳、断线指数退避重连并 RESUME 续传；服务端约 1 小时会断开连接，SDK 会在临到期前（约 50 分钟）主动重连续传 | SDK 处理 Ed25519 验签、op=13 验证应答、60s TTL 事件去重 |
|
||||
| 适用 | 长驻进程、需要会话与重连保障 | 无状态部署、回调型服务 |
|
||||
|
||||
两种方式下 handler 的写法完全一致，切换只需改 `CONNECTER`。
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user