From b37a53f6e0dec360a3e684fbafd7140ec00a2711 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Mon, 21 Sep 2026 18:25:16 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=87=8D=E8=BF=9E=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E7=AD=89=E5=BE=85=E5=B9=B6=E7=B2=BE=E7=AE=80=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=E6=97=A5=E5=BF=97=EF=BC=8C=E5=A4=B1=E8=B4=A5=E6=89=8D=E8=AE=B0?= =?UTF-8?q?=E6=96=AD=E5=BC=80=E4=B8=8E=E9=80=80=E9=81=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 连接曾稳定存活（>= _STABLE_SECONDS）而被服务端要求重连时立刻重连、 不等待也不记日志：这期间事件由网关按 RESUME 补发，等待只是徒增盲区 - 仅在连接未能稳定存活（握手失败/看门狗超时/限流/抖动）时才记 "WebSocket 已断开"并按指数退避重试 - 去掉常规的「WebSocket 已连接/已断开」info，只保留「发送 RESUME」等 有信息量的行，消除每小时一次的日志噪音 - 新增 tests/test_reconnect.py：健康断开零等待零日志、失败退避翻倍、 限流 60s、抖动不重置退避 --- src/qqbotsdk/connecter.py | 34 ++++++++++---- tests/test_reconnect.py | 93 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+), 8 deletions(-) create mode 100644 tests/test_reconnect.py diff --git a/src/qqbotsdk/connecter.py b/src/qqbotsdk/connecter.py index 1b71853..5f9802f 100644 --- a/src/qqbotsdk/connecter.py +++ b/src/qqbotsdk/connecter.py @@ -72,22 +72,42 @@ class WebsocketConnecter(Connecter): self._outbound: asyncio.Queue[Payload] = asyncio.Queue() async def run(self) -> None: + """重连主循环。 + + 健康连接被服务端要求重连（如满 1 小时的连接限寿）时不等待、也不记 + 连接/断开日志，直接立刻重连——这期间的事件由网关按 RESUME 补发， + 盲区最小；只有连接未能稳定存活（握手失败、看门狗超时、限流、抖动） + 才记断开并按退避等待，`_STABLE_SECONDS` 保证抖动的服务端不会被 1s + 间隔反复探测。 + """ backoff: float = 1 + delay: float = 0 # 0 表示立刻重连（首次重试不等待） while True: + if delay: + logger.info(f"{delay}s 后重连") + await asyncio.sleep(delay) try: lived = await self._connect() - if lived >= self._STABLE_SECONDS: - backoff = 1 except RateLimitError: logger.warning("gateway 触发频率限制") - backoff = 60 + delay = backoff = 60 + continue except TimeoutError: logger.warning("心跳超时，触发看门狗重连") + lived = 0.0 except (aiohttp.ClientError, RuntimeError) as e: logger.warning(f"WebSocket 连接异常: {e}") - logger.info(f"{backoff}s 后重连") - await asyncio.sleep(backoff) - backoff = min(backoff * 2, 60) + lived = 0.0 + + if lived >= self._STABLE_SECONDS: + # 连接曾稳定存活：属正常断开（服务端要求重连），立刻重连 + backoff = 1 + delay = 0 + else: + # 没能稳定存活：需要延迟重试，此时才记断开 + logger.warning("WebSocket 已断开") + delay = backoff + backoff = min(backoff * 2, 60) async def _connect(self) -> float: """建立一次连接直到断开，返回存活秒数（供 run 判断是否重置退避）。""" @@ -107,7 +127,6 @@ class WebsocketConnecter(Connecter): raise RuntimeError(f"获取 gateway 失败: {data}") async with self._http.ws_connect(gateway, headers=headers) as ws: - logger.info("WebSocket 已连接") # HELLO 才给出心跳间隔；receive 解析到后经 Future 交给心跳任务 interval: asyncio.Future[float] = asyncio.get_running_loop().create_future() receive = asyncio.create_task(self.receive_helper(ws, interval)) @@ -121,7 +140,6 @@ class WebsocketConnecter(Connecter): # 取回被取消任务的结果，避免 "Task exception was never retrieved"， # 也确保它们在下一轮重连前真正停下 await asyncio.gather(send, heartbeat, return_exceptions=True) - logger.info("WebSocket 已断开") return monotonic() - started async def receive_helper( diff --git a/tests/test_reconnect.py b/tests/test_reconnect.py new file mode 100644 index 0000000..fdf8cf2 --- /dev/null +++ b/tests/test_reconnect.py @@ -0,0 +1,93 @@ +"""重连循环：健康断开立刻重连（不记连接/断开），失败才退避并记断开。""" + +import asyncio +from typing import Any + +import pytest +from loguru import logger + +from qqbotsdk.connecter import RateLimitError, WebsocketConnecter + +pytestmark = pytest.mark.asyncio + + +def _make(monkeypatch: pytest.MonkeyPatch, seq: list[Any]) -> WebsocketConnecter: + """构造 connecter，并把 _connect 换成按 seq 依次返回（float=存活秒数， + Exception=连接失败）。seq 用尽后抛 IndexError 中止 run() 的无限循环。""" + conn = WebsocketConnecter.__new__(WebsocketConnecter) + remaining = list(seq) + + async def fake_connect() -> float: + item = remaining.pop(0) # 用尽则 IndexError 逃出 run() + if isinstance(item, Exception): + raise item + return float(item) + + monkeypatch.setattr(conn, "_connect", fake_connect) + return conn + + +async def _run_capture( + monkeypatch: pytest.MonkeyPatch, seq: list[Any] +) -> tuple[list[float], list[str]]: + """跑 run() 直到 seq 用尽，返回 (sleep 延迟序列, 日志消息列表)。""" + sleeps: list[float] = [] + logs: list[str] = [] + real_sleep = asyncio.sleep + + async def fake_sleep(delay: float) -> None: + sleeps.append(delay) + await real_sleep(0) # 让出控制权，避免紧循环 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + sink_id = logger.add(lambda m: logs.append(m.record["message"]), level="INFO") + try: + conn = _make(monkeypatch, seq) + with pytest.raises(IndexError): + await conn.run() + finally: + logger.remove(sink_id) + return sleeps, logs + + +async def test_healthy_disconnect_reconnects_without_delay_or_log( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """健康连接（存活 >= 稳定阈值）断开后立刻重连：不 sleep、不记连接/断开。""" + sleeps, logs = await _run_capture(monkeypatch, [3600.0, 3600.0]) + + assert sleeps == [] # 两次都立刻重连 + assert not any("已连接" in m or "已断开" in m for m in logs) + + +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] + ) + + # 健康断开 → 无 sleep；失败后依次 1s、2s、4s + assert sleeps == [1, 2, 4] + assert sum("已断开" in m for m in logs) == 3 + + +async def test_rate_limit_backs_off_hard( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """gateway 限流：退避 60s，且不重复记断开。""" + sleeps, logs = await _run_capture(monkeypatch, [RateLimitError(), 3600.0]) + + assert sleeps == [60] + assert any("频率限制" in m for m in logs) + assert not any("已断开" in m for m in logs) + + +async def test_flapping_connection_does_not_reset_backoff( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """连接建立后立刻掉线（存活 < 阈值）视为抖动：退避持续拉长，不归零。""" + sleeps, _ = await _run_capture(monkeypatch, [0.5, 0.5, 0.5]) + + assert sleeps == [1, 2, 4] # 每次都等得更久（若归零会是 1,1,1）