diff --git a/app/modules/qqbot/gateway.py b/app/modules/qqbot/gateway.py index 7fe76bd14..faa61cf2b 100644 --- a/app/modules/qqbot/gateway.py +++ b/app/modules/qqbot/gateway.py @@ -39,27 +39,116 @@ def run_gateway( :param ws_holder: 调用方持有的单元素列表，存放当前 WebSocketApp，供 stop() 时 close 以打断 run_forever """ last_seq: Optional[int] = None - heartbeat_interval_ms: Optional[int] = None - heartbeat_timer: Optional[threading.Timer] = None + heartbeat_thread: Optional[threading.Thread] = None + heartbeat_stop_event: Optional[threading.Event] = None + heartbeat_epoch = 0 + heartbeat_state_lock = threading.Lock() + heartbeat_replace_lock = threading.RLock() - def send_heartbeat(): - nonlocal heartbeat_timer - if stop_event.is_set(): + def heartbeat_loop( + generation_epoch: int, + generation_stop_event: threading.Event, + interval_seconds: float, + ws: websocket.WebSocketApp, + ) -> None: + """在当前连接 generation 内串行发送心跳，停止后不再派生新线程。""" + while not generation_stop_event.wait(interval_seconds): + if stop_event.is_set(): + return + with heartbeat_state_lock: + if ( + generation_epoch != heartbeat_epoch + or heartbeat_stop_event is not generation_stop_event + ): + return + sequence = last_seq + try: + payload = {"op": 1, "d": sequence} + ws.send(json.dumps(payload)) + logger.debug( + f"[QQ Gateway:{config_name}] Heartbeat sent, seq={sequence}" + ) + except Exception as err: + logger.debug(f"[QQ Gateway:{config_name}] Heartbeat error: {err}") + + def invalidate_heartbeat_generation() -> tuple[Optional[threading.Thread], int]: + """锁内失效当前 generation，并返回待等待的 owner 与新 epoch。""" + nonlocal heartbeat_epoch + with heartbeat_state_lock: + heartbeat_epoch += 1 + current_thread = heartbeat_thread + if heartbeat_stop_event is not None: + heartbeat_stop_event.set() + return current_thread, heartbeat_epoch + + def clear_heartbeat_generation(current_thread: Optional[threading.Thread]) -> None: + """仅在快照仍是当前 owner 且已经终止时清除 generation 引用。""" + nonlocal heartbeat_thread, heartbeat_stop_event + if current_thread is not None and current_thread.is_alive(): return - try: - if ws_holder and ws_holder[0]: - payload = {"op": 1, "d": last_seq} - ws_holder[0].send(json.dumps(payload)) - logger.debug(f"[QQ Gateway:{config_name}] Heartbeat sent, seq={last_seq}") - except Exception as err: - logger.debug(f"[QQ Gateway:{config_name}] Heartbeat error: {err}") - if heartbeat_interval_ms and not stop_event.is_set(): - heartbeat_timer = threading.Timer(heartbeat_interval_ms / 1000.0, send_heartbeat) - heartbeat_timer.daemon = True - heartbeat_timer.start() + with heartbeat_state_lock: + if heartbeat_thread is current_thread: + heartbeat_thread = None + heartbeat_stop_event = None - def on_ws_message(_, message): - nonlocal last_seq, heartbeat_interval_ms, heartbeat_timer + def stop_heartbeat(*, wait: bool) -> None: + """停止当前心跳；等待阶段不持有状态锁，避免阻塞 close 回调。""" + if wait: + heartbeat_replace_lock.acquire() + try: + current_thread, _ = invalidate_heartbeat_generation() + if ( + wait + and current_thread is not None + and current_thread.is_alive() + and current_thread is not threading.current_thread() + ): + current_thread.join() + clear_heartbeat_generation(current_thread) + finally: + if wait: + heartbeat_replace_lock.release() + + def start_heartbeat(ws: websocket.WebSocketApp, interval_ms: int) -> None: + """停止旧 generation 后启动一个由 Gateway 聚合的新心跳 owner。""" + nonlocal heartbeat_thread, heartbeat_stop_event + with heartbeat_replace_lock: + current_thread, generation_epoch = invalidate_heartbeat_generation() + if ( + current_thread is not None + and current_thread.is_alive() + and current_thread is not threading.current_thread() + ): + current_thread.join() + clear_heartbeat_generation(current_thread) + with heartbeat_state_lock: + if ( + generation_epoch != heartbeat_epoch + or stop_event.is_set() + or not ws_holder + or ws_holder[0] is not ws + ): + return + generation_stop_event = threading.Event() + # 心跳是 Gateway 的常驻子 owner，不占用进程共享 ThreadHelper；Gateway 退出前统一 join。 + generation_thread = threading.Thread( + target=heartbeat_loop, + args=( + generation_epoch, + generation_stop_event, + interval_ms / 1000.0, + ws, + ), + daemon=True, + name=f"qq-gateway-heartbeat-{config_name}", + ) + heartbeat_stop_event = generation_stop_event + heartbeat_thread = generation_thread + generation_thread.start() + + def on_ws_message(ws, message): + """处理当前 WebSocket 连接收到的 QQ Gateway 消息。""" + nonlocal last_seq try: payload = json.loads(message) except json.JSONDecodeError as err: @@ -72,7 +161,8 @@ def run_gateway( t = payload.get("t") if s is not None: - last_seq = s + with heartbeat_state_lock: + last_seq = s logger.debug(f"[QQ Gateway:{config_name}] op={op} t={t}") @@ -89,15 +179,10 @@ def run_gateway( "shard": [0, 1], }, } - ws_holder[0].send(json.dumps(identify)) + ws.send(json.dumps(identify)) logger.info(f"[QQ Gateway:{config_name}] Identify sent") - # 启动心跳 - if heartbeat_timer: - heartbeat_timer.cancel() - heartbeat_timer = threading.Timer(heartbeat_interval_ms / 1000.0, send_heartbeat) - heartbeat_timer.daemon = True - heartbeat_timer.start() + start_heartbeat(ws, heartbeat_interval_ms) elif op == 0: # Dispatch if t == "READY": @@ -147,59 +232,69 @@ def run_gateway( elif op == 9: # Invalid Session logger.warning(f"[QQ Gateway:{config_name}] Invalid session") - if ws_holder and ws_holder[0]: - ws_holder[0].close() + ws.close() def on_ws_error(_, error): + """记录当前 WebSocket 连接上报的错误。""" logger.error(f"[QQ Gateway:{config_name}] WebSocket error: {error}") - def on_ws_close(_, close_status_code, close_msg): + def on_ws_close(ws, close_status_code, close_msg): + """失效当前连接及其心跳 generation，但不在回调线程等待。""" logger.info(f"[QQ Gateway:{config_name}] WebSocket closed: {close_status_code} {close_msg}") - if heartbeat_timer: - heartbeat_timer.cancel() - ws_holder.clear() + # close 回调可能由 stop() 调用线程触发，只发信号，统一由 Gateway 线程等待终态。 + with heartbeat_state_lock: + is_current_connection = bool(ws_holder and ws_holder[0] is ws) + if is_current_connection: + ws_holder.clear() + if is_current_connection: + stop_heartbeat(wait=False) reconnect_delays = [1, 2, 5, 10, 30, 60] attempt = 0 - while not stop_event.is_set(): - try: - token = get_token_fn(app_id, app_secret) - gateway_url = get_gateway_url_fn(token) - logger.info(f"[QQ Gateway:{config_name}] Connecting to {gateway_url[:60]}...") + try: + while not stop_event.is_set(): + try: + token = get_token_fn(app_id, app_secret) + gateway_url = get_gateway_url_fn(token) + logger.info(f"[QQ Gateway:{config_name}] Connecting to {gateway_url[:60]}...") - ws = websocket.WebSocketApp( - gateway_url, - on_message=on_ws_message, - on_error=on_ws_error, - on_close=on_ws_close, - ) - ws_holder.clear() - ws_holder.append(ws) + ws = websocket.WebSocketApp( + gateway_url, + on_message=on_ws_message, + on_error=on_ws_error, + on_close=on_ws_close, + ) + with heartbeat_state_lock: + ws_holder.clear() + ws_holder.append(ws) - # run_forever 会阻塞，需要传入 stop_event 的检查 - # websocket-client 的 run_forever 支持 ping_interval, ping_timeout - # 我们使用自定义心跳，所以不设置 ping - ws.run_forever( - ping_interval=None, - ping_timeout=None, - skip_utf8_validation=True, - ) + # websocket-client 的 run_forever 会阻塞；QQ 协议使用自定义心跳，不启用 ping。 + try: + ws.run_forever( + ping_interval=None, + ping_timeout=None, + skip_utf8_validation=True, + ) + finally: + # 旧连接的心跳必须先终止，下一轮 reconnect 才能取得 owner。 + stop_heartbeat(wait=True) - except Exception as e: - logger.error(f"[QQ Gateway:{config_name}] Connection error: {e}") + except Exception as err: + logger.error(f"[QQ Gateway:{config_name}] Connection error: {err}") - if stop_event.is_set(): - break - - delay = reconnect_delays[min(attempt, len(reconnect_delays) - 1)] - attempt += 1 - logger.info(f"[QQ Gateway:{config_name}] Reconnecting in {delay}s (attempt {attempt})") - for _ in range(delay * 10): if stop_event.is_set(): break - time.sleep(0.1) - if heartbeat_timer: - heartbeat_timer.cancel() - logger.info(f"[QQ Gateway:{config_name}] Gateway thread stopped") + delay = reconnect_delays[min(attempt, len(reconnect_delays) - 1)] + attempt += 1 + logger.info( + f"[QQ Gateway:{config_name}] Reconnecting in {delay}s (attempt {attempt})" + ) + for _ in range(delay * 10): + if stop_event.is_set(): + break + time.sleep(0.1) + finally: + stop_heartbeat(wait=True) + logger.info(f"[QQ Gateway:{config_name}] Gateway thread stopped") diff --git a/docs/refactor/backend-architecture-review.md b/docs/refactor/backend-architecture-review.md index 355d4a702..d2373e141 100644 --- a/docs/refactor/backend-architecture-review.md +++ b/docs/refactor/backend-architecture-review.md @@ -85,9 +85,9 @@ |---|---|---|---| | 0 | 历史任务清账、现行架构图、外部契约核对和宿主基线对齐 | 已推送 | `d234c7132`；远端同 SHA；ahead/behind `0/0`；架构契约 `71 passed` | | 1 | Mypy fail-closed，并把 Ruff/Mypy 已下降债务固化为真实低水位 | 已推送 | `6062b0661`；远端同 SHA；ahead/behind `0/0`；Mypy 11994、Ruff 976 | -| 2 | 用全量串行测试初始化非零 Coverage 低水位，并补齐 CI/文档防回退契约 | 已本地验证 | Ubuntu canonical：Application `9292/11949`（77.76%），Domain `3390/4278`（79.24%）；专项 `26 passed` | -| 3 | 收口阶段 62 遗留的 QQ Gateway heartbeat Timer 所有权 | 待批次 2 | Timer 只 cancel 不 join，Gateway 主线程可能在 heartbeat 仍执行时报告停止成功 | -| Final | 全仓回归、插件兼容复核、台账定稿和远端一致性验证 | 待前置批次 | 所有准入项已推送；全量测试和适用门禁通过；本地/远端 0/0 | +| 2 | 用全量串行测试初始化非零 Coverage 低水位，并补齐 CI/文档防回退契约 | 已推送 | `265d3c6d1`；远端同 SHA；ahead/behind `0/0`；Application 77.76%，Domain 79.24% | +| 3 | 收口阶段 62 遗留的 QQ Gateway heartbeat Timer 所有权 | 已全量验证 | generation owner 已实现；全量 `6391 passed, 6 skipped`；等待提交推送 | +| Final | 全仓回归、插件兼容复核、台账定稿和远端一致性验证 | 进行中 | 本地门禁已通过；等待批次 3 推送及远端 0/0 复核 | ### 批次 0：审计与基线对齐 @@ -162,7 +162,8 @@ 最新专项覆盖，质量/CI 专项 `26 passed`，Ruff 通过，Pylint `10.00/10`，宿主、Schema、 Coverage 和锁文件门禁均通过。 -待完成：提交推送并记录远端证据。 +交付证据：提交 `265d3c6d1` 已推送到 `origin/v3`；`git ls-remote` 返回同一 SHA， +`HEAD...origin/v3` 为 `0/0`，提交严格包含批次 2 的 8 个主仓路径。 ### 批次 3：QQ Gateway heartbeat owner @@ -171,11 +172,29 @@ WebSocket 和 Gateway 退出路径只调用 `cancel()`，而 `QQBot.stop()` 只等待 Gateway 主线程。 当 heartbeat callback 已进入 `send()` 时，`cancel()` 不能终止它，外层因此可能误报收敛。 -停止条件：Gateway 最终退出路径在现有外层 20 秒硬预算内等待当前 Timer 真正终止； -Timer 未终止时 Gateway 线程保持存活，`QQBot` 保留 owner 并允许再次停止；故障注入证明首次 +停止条件：Gateway 最终退出路径等待当前 heartbeat owner 真正终止，不为 heartbeat 叠加 +第二份调用方等待预算；heartbeat 未终止时 Gateway 线程保持存活，`QQBot` 在现有 20 秒 +Gateway join 预算后保留 owner 并允许再次停止；故障注入证明首次 停止返回 `False`、释放 callback 后第二次返回 `True`；生命周期专项和 task ownership 门禁通过； 批次独立提交推送。 +本地实现与专项验收： + +* 递归 Timer 已替换为一个由 Gateway 聚合、按 connection generation 捕获精确 WebSocket 的 + 常驻心跳线程；新 Hello 必须先失效并等待旧 generation，旧连接 close 不能终止新连接心跳； +* 状态锁只保护 epoch、owner 和 sequence 快照，任何 `join()`、网络发送及 close 回调都不持锁； + Gateway 的每次连接和最终退出都等待子 owner 终止，所以未收敛时外层 Gateway 线程保持存活； +* 回调中的 Identify、Invalid Session 均使用回调携带的精确 WebSocket，避免 `ws_holder` + 被并发 close 清空或替换后误发到别的连接；公开 `QQBot.stop()` 合同和等待预算未改变； +* 三条阻塞发送、重复 Hello 和双连接重连故障注入测试并行重复 4 轮均通过；完整生命周期专项 + `96 passed`，Ruff 通过，Pylint `10.00/10`，task ownership、复杂度、async blocking、 + 宿主基线及 Ruff/Mypy ratchet 均通过；Mypy 低水位从 11994 净减至 11993。 +* 双连接用例直接证明重连前必须等待旧心跳终止，且迟到的旧连接 close 不会终止新 generation； + 独立代码审查未发现生产线程生命周期、锁序、generation 或 reconnect 合同缺陷； +* 4 分片全量回归 `6391 passed, 6 skipped`；锁文件、Schema、Coverage、service locator 和 + 官方插件 ABI 语义门禁均通过，架构/质量专项 `50 passed`。插件参考仓只有不参与语义判定的 + schema/provenance 漂移，且本地分支落后其远端 2 个提交，本轮保持参考仓只读且不刷新指纹。 + ## 五、验证矩阵 每个批次按改动范围选择下列命令，Final 全部执行： @@ -217,6 +236,10 @@ git rev-list --left-right --count HEAD...origin/v3 | 2026-08-26 | 批次 1 交付 | `6062b0661` 已推送；远端同 SHA；ahead/behind `0/0`；显式变更 10 个路径 | | 2026-08-26 | 批次 2 canonical Coverage | run `32977180133` 与本机计数一致；Application 77.76%，Domain 79.24%；零 fixture 已初始化 | | 2026-08-26 | 批次 2 本地验收 | 串行全量 `6379 passed, 6 skipped`；最新专项 `26 passed`；Coverage/宿主/Schema/锁文件门禁通过 | +| 2026-08-26 | 批次 2 交付 | `265d3c6d1` 已推送；远端同 SHA；ahead/behind `0/0`；显式变更 8 个路径 | +| 2026-08-26 | 批次 3 启动 | 仅收口阶段 62 已记录的 QQ heartbeat owner，不扩张公开 `QQBot.stop()` 合同 | +| 2026-08-26 | 批次 3 专项验收 | 三条故障注入并行重复 4 轮稳定；生命周期 `96 passed`；静态、所有权和架构门禁通过 | +| 2026-08-26 | 批次 3 全量验收 | 最终快照 4 分片 `6391 passed, 6 skipped`；全部适用主仓门禁通过；插件 ABI 语义无变化 | ## 七、本轮停止条件 diff --git a/docs/rules/06-code-styles.md b/docs/rules/06-code-styles.md index 459d6e84f..3bd19bf18 100644 --- a/docs/rules/06-code-styles.md +++ b/docs/rules/06-code-styles.md @@ -35,7 +35,10 @@ - Prefer `async def` for I/O-bound operations (network requests, database queries, file operations). - Use `await` consistently; do not mix sync and async code paths in the same function without using `run_in_threadpool` from FastAPI or `asyncio.to_thread`. - For CPU-bound work that must not block the event loop, submit to `ThreadHelper` (see `app/runtime/thread.py`). -- Do not use bare `threading.Thread` in new code; use `ThreadHelper.submit()`. +- Use `ThreadHelper.submit()` for finite background work. A long-lived protocol loop may use a + dedicated `threading.Thread` only when its existing lifecycle owner signals and joins that child, + while the external caller bounded-waits and retains the parent owner on non-convergence; do not + create a dedicated thread pool for that exception. --- diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 7f88e844e..c0dd84e93 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -2580,7 +2580,7 @@ "type-arg": 13 }, "app/modules/qqbot/gateway.py": { - "no-untyped-def": 4, + "no-untyped-def": 3, "type-arg": 2 }, "app/modules/qqbot/module.py": { diff --git a/tests/test_module_lifecycle.py b/tests/test_module_lifecycle.py index 491b6be3c..d86d88d04 100644 --- a/tests/test_module_lifecycle.py +++ b/tests/test_module_lifecycle.py @@ -1,4 +1,5 @@ import asyncio +import json import threading import time from unittest.mock import Mock, patch @@ -13,6 +14,7 @@ from app.modules.feishu import FeishuModule from app.modules.feishu.feishu import Feishu from app.modules.filter import FilterModule from app.modules.plex import PlexModule +from app.modules.qqbot import gateway as qq_gateway from app.modules.qqbot.module import QQBotModule from app.modules.qqbot.qqbot import QQBot from app.modules.slack import SlackModule @@ -328,6 +330,335 @@ def test_qqbot_stop_retains_gateway_thread_until_retry() -> None: assert client.stop() is True +def test_qqbot_stop_retains_gateway_until_inflight_heartbeat_finishes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """心跳发送仍阻塞时 Gateway 必须保持存活，释放后才允许关闭成功。""" + heartbeat_started = threading.Event() + release_heartbeat = threading.Event() + connection_closed = threading.Event() + callbacks = {} + fake_ws = Mock() + hello_payload = json.dumps({"op": 10, "d": {"heartbeat_interval": 1}}) + + def build_websocket(_url, **kwargs): + """保存 Gateway 回调并返回隔离的 WebSocket 桩。""" + callbacks.update(kwargs) + return fake_ws + + def send(payload: str) -> None: + """Identify 立即完成，首个心跳保持在发送中的故障状态。""" + if json.loads(payload).get("op") == 1: + heartbeat_started.set() + release_heartbeat.wait() + + def close() -> None: + """同步触发 close 回调，复现 QQBot.stop() 的真实调用线程。""" + callbacks["on_close"](fake_ws, 1000, "test close") + connection_closed.set() + + def run_forever(**_kwargs) -> None: + """发送 Hello 后保持连接，直到 stop() 主动关闭。""" + callbacks["on_message"](fake_ws, hello_payload) + connection_closed.wait() + + fake_ws.send.side_effect = send + fake_ws.close.side_effect = close + fake_ws.run_forever.side_effect = run_forever + monkeypatch.setattr(qq_gateway.websocket, "WebSocketApp", build_websocket) + + client = QQBot.__new__(QQBot) + client._gateway_stop = threading.Event() + client._gateway_ws_holder = [] + client._gateway_join_timeout_seconds = 0.02 + gateway_thread = threading.Thread( + target=qq_gateway.run_gateway, + kwargs={ + "app_id": "app-id", + "app_secret": "secret", + "config_name": "test", + "get_token_fn": lambda _app_id, _secret: "token", + "get_gateway_url_fn": lambda _token: "wss://gateway.test", + "on_message_fn": lambda _payload: None, + "stop_event": client._gateway_stop, + "ws_holder": client._gateway_ws_holder, + }, + daemon=True, + ) + client._gateway_thread = gateway_thread + gateway_thread.start() + + try: + assert heartbeat_started.wait(0.5) + started_at = time.monotonic() + assert client.stop() is False + assert time.monotonic() - started_at < 0.2 + assert gateway_thread.is_alive() + + release_heartbeat.set() + gateway_thread.join(timeout=1.0) + assert client.stop() is True + assert not gateway_thread.is_alive() + finally: + client._gateway_stop.set() + release_heartbeat.set() + connection_closed.set() + gateway_thread.join(timeout=1.0) + + +def test_qq_gateway_replaces_heartbeat_generation_without_overlap( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """重复 Hello 必须等待旧心跳终止，任何时刻只能有一个发送 generation。""" + first_heartbeat_started = threading.Event() + release_first_heartbeat = threading.Event() + second_heartbeat_started = threading.Event() + release_second_heartbeat = threading.Event() + third_heartbeat_started = threading.Event() + second_hello_entered = threading.Event() + third_hello_entered = threading.Event() + connection_ready = threading.Event() + connection_closed = threading.Event() + counters_lock = threading.Lock() + callbacks = {} + fake_ws = Mock() + hello_payload = json.dumps({"op": 10, "d": {"heartbeat_interval": 1}}) + heartbeat_count = 0 + active_heartbeats = 0 + max_active_heartbeats = 0 + + def build_websocket(_url, **kwargs): + """保存 Gateway 回调并返回隔离的 WebSocket 桩。""" + callbacks.update(kwargs) + return fake_ws + + def send(payload: str) -> None: + """阻塞第一代心跳，并记录是否出现跨 generation 并发发送。""" + nonlocal heartbeat_count, active_heartbeats, max_active_heartbeats + if json.loads(payload).get("op") != 1: + return + with counters_lock: + heartbeat_count += 1 + current_heartbeat = heartbeat_count + active_heartbeats += 1 + max_active_heartbeats = max( + max_active_heartbeats, + active_heartbeats, + ) + try: + if current_heartbeat == 1: + first_heartbeat_started.set() + release_first_heartbeat.wait() + elif current_heartbeat == 2: + second_heartbeat_started.set() + release_second_heartbeat.wait() + elif current_heartbeat == 3: + third_heartbeat_started.set() + finally: + with counters_lock: + active_heartbeats -= 1 + + def run_forever(**_kwargs) -> None: + """发送首个 Hello 后保持连接，第二个 Hello 由测试线程注入。""" + callbacks["on_message"](fake_ws, hello_payload) + connection_ready.set() + connection_closed.wait() + + def send_second_hello() -> None: + """从独立调用线程注入重复 Hello，以观测 generation 屏障。""" + second_hello_entered.set() + callbacks["on_message"](fake_ws, hello_payload) + + def send_third_hello() -> None: + """在第二代发送中再次注入 Hello，供 close 失效待发布 generation。""" + third_hello_entered.set() + callbacks["on_message"](fake_ws, hello_payload) + + fake_ws.send.side_effect = send + fake_ws.run_forever.side_effect = run_forever + monkeypatch.setattr(qq_gateway.websocket, "WebSocketApp", build_websocket) + + stop_event = threading.Event() + ws_holder = [] + gateway_thread = threading.Thread( + target=qq_gateway.run_gateway, + kwargs={ + "app_id": "app-id", + "app_secret": "secret", + "config_name": "generation-test", + "get_token_fn": lambda _app_id, _secret: "token", + "get_gateway_url_fn": lambda _token: "wss://gateway.test", + "on_message_fn": lambda _payload: None, + "stop_event": stop_event, + "ws_holder": ws_holder, + }, + daemon=True, + ) + gateway_thread.start() + hello_thread = threading.Thread(target=send_second_hello, daemon=True) + hello_thread_started = False + third_hello_thread = threading.Thread(target=send_third_hello, daemon=True) + third_hello_thread_started = False + + try: + assert connection_ready.wait(0.5) + assert first_heartbeat_started.wait(0.5) + hello_thread.start() + hello_thread_started = True + assert second_hello_entered.wait(0.2) + assert not second_heartbeat_started.wait(0.03) + assert hello_thread.is_alive() + assert max_active_heartbeats == 1 + + release_first_heartbeat.set() + hello_thread.join(timeout=1.0) + assert not hello_thread.is_alive() + assert second_heartbeat_started.wait(0.5) + assert max_active_heartbeats == 1 + + third_hello_thread.start() + third_hello_thread_started = True + assert third_hello_entered.wait(0.2) + assert third_hello_thread.is_alive() + close_started_at = time.monotonic() + callbacks["on_close"](fake_ws, 1000, "close during replacement") + assert time.monotonic() - close_started_at < 0.2 + + release_second_heartbeat.set() + third_hello_thread.join(timeout=1.0) + assert not third_hello_thread.is_alive() + assert not third_heartbeat_started.wait(0.05) + assert max_active_heartbeats == 1 + + stop_event.set() + connection_closed.set() + gateway_thread.join(timeout=1.0) + assert not gateway_thread.is_alive() + finally: + stop_event.set() + release_first_heartbeat.set() + release_second_heartbeat.set() + connection_closed.set() + if hello_thread_started: + hello_thread.join(timeout=1.0) + if third_hello_thread_started: + third_hello_thread.join(timeout=1.0) + gateway_thread.join(timeout=1.0) + + +def test_qq_gateway_reconnect_joins_old_heartbeat_and_ignores_stale_close( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """重连必须先回收旧心跳，迟到的旧连接 close 不得终止新 generation。""" + first_callbacks = {} + second_callbacks = {} + first_ws = Mock() + second_ws = Mock() + first_heartbeat_started = threading.Event() + release_first_heartbeat = threading.Event() + return_first_connection = threading.Event() + second_connection_created = threading.Event() + second_heartbeat_started = threading.Event() + release_second_heartbeat = threading.Event() + second_followup_started = threading.Event() + return_second_connection = threading.Event() + hello_payload = json.dumps({"op": 10, "d": {"heartbeat_interval": 1}}) + websocket_count = 0 + second_heartbeat_count = 0 + + def build_websocket(_url, **kwargs): + """依次创建两条连接，并保留各自回调以注入迟到 close。""" + nonlocal websocket_count + websocket_count += 1 + if websocket_count == 1: + first_callbacks.update(kwargs) + return first_ws + if websocket_count == 2: + second_callbacks.update(kwargs) + second_connection_created.set() + return second_ws + raise AssertionError("Gateway 在停止后不应建立第三条测试连接") + + def send_first(payload: str) -> None: + """阻塞第一条连接的心跳，供重连屏障检查。""" + if json.loads(payload).get("op") == 1: + first_heartbeat_started.set() + release_first_heartbeat.wait() + + def send_second(payload: str) -> None: + """阻塞新连接首个心跳，并记录迟到 close 后是否继续工作。""" + nonlocal second_heartbeat_count + if json.loads(payload).get("op") != 1: + return + second_heartbeat_count += 1 + if second_heartbeat_count == 1: + second_heartbeat_started.set() + release_second_heartbeat.wait() + else: + second_followup_started.set() + + def run_first(**_kwargs) -> None: + """第一条连接握手后按测试信号返回，触发真实 reconnect 路径。""" + first_callbacks["on_message"](first_ws, hello_payload) + return_first_connection.wait() + + def run_second(**_kwargs) -> None: + """第二条连接保持运行，直到测试完成迟到 close 校验。""" + second_callbacks["on_message"](second_ws, hello_payload) + return_second_connection.wait() + + first_ws.send.side_effect = send_first + first_ws.run_forever.side_effect = run_first + second_ws.send.side_effect = send_second + second_ws.run_forever.side_effect = run_second + monkeypatch.setattr(qq_gateway.websocket, "WebSocketApp", build_websocket) + monkeypatch.setattr(qq_gateway.time, "sleep", lambda _seconds: None) + + stop_event = threading.Event() + ws_holder = [] + gateway_thread = threading.Thread( + target=qq_gateway.run_gateway, + kwargs={ + "app_id": "app-id", + "app_secret": "secret", + "config_name": "reconnect-test", + "get_token_fn": lambda _app_id, _secret: "token", + "get_gateway_url_fn": lambda _token: "wss://gateway.test", + "on_message_fn": lambda _payload: None, + "stop_event": stop_event, + "ws_holder": ws_holder, + }, + daemon=True, + ) + gateway_thread.start() + + try: + assert first_heartbeat_started.wait(0.5) + return_first_connection.set() + assert not second_connection_created.wait(0.05) + + release_first_heartbeat.set() + assert second_connection_created.wait(0.5) + assert second_heartbeat_started.wait(0.5) + + first_callbacks["on_close"](first_ws, 1000, "stale close") + release_second_heartbeat.set() + assert second_followup_started.wait(0.5) + + stop_event.set() + second_callbacks["on_close"](second_ws, 1000, "current close") + return_second_connection.set() + gateway_thread.join(timeout=1.0) + assert not gateway_thread.is_alive() + finally: + stop_event.set() + release_first_heartbeat.set() + release_second_heartbeat.set() + return_first_connection.set() + return_second_connection.set() + gateway_thread.join(timeout=1.0) + + def test_wechat_bot_stop_reports_each_live_thread_until_retry() -> None: """企业微信网关或心跳任一存活时都必须返回未收敛。""" client = WeChatBot.__new__(WeChatBot)