From be7dfd77a337e3bd593bd990e26b2450da064837 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Tue, 25 Aug 2026 06:58:31 +0800 Subject: [PATCH] =?UTF-8?q?fix(runtime):=20=E6=94=B6=E6=95=9B=E6=A8=A1?= =?UTF-8?q?=E5=9D=97=E5=85=B3=E9=97=AD=E7=BA=BF=E7=A8=8B=E6=89=80=E6=9C=89?= =?UTF-8?q?=E6=9D=83=20(#6443)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(runtime): bound module shutdown owners * fix(runtime): declare blocking shutdown owners --------- Co-authored-by: jxxghp --- app/modules/telegram/telegram.py | 32 +++- app/startup/initializers/modules.py | 23 ++- app/startup/lifecycle/__init__.py | 76 +++++--- scripts/perf/README.md | 16 ++ scripts/perf/module_shutdown_ab.py | 179 ++++++++++++++++++ .../architecture/dependency-baseline.json | 6 +- tests/test_lifecycle_shutdown.py | 72 ++++++- tests/test_module_lifecycle.py | 65 ++++++- 8 files changed, 421 insertions(+), 48 deletions(-) create mode 100644 scripts/perf/module_shutdown_ab.py diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index 504e44d03..4b24d8376 100644 --- a/app/modules/telegram/telegram.py +++ b/app/modules/telegram/telegram.py @@ -84,7 +84,7 @@ class Telegram: _typing_command_max_duration_seconds = 30 _typing_callback_max_duration_seconds = 60 _typing_join_timeout_seconds = 1 - _polling_join_timeout_seconds = 10 + _shutdown_timeout_seconds = 10 def __init__( self, @@ -1743,9 +1743,25 @@ class Telegram: # 清理菜单命令 self._bot.delete_my_commands() + @staticmethod + def _stop_bot_with_deadline(bot: TeleBot, deadline: float) -> bool: + """停止 SDK polling，并在共享 deadline 内等待 worker 收敛。""" + bot.stop_polling() + if not bot.threaded or not bot.worker_pool: + return True + + workers = tuple(bot.worker_pool.workers) + for worker in workers: + worker.stop() + for worker in workers: + if worker is threading.current_thread(): + continue + worker.join(timeout=max(0.0, deadline - time.monotonic())) + return all(not worker.is_alive() for worker in workers) + def stop(self) -> bool: """ - 停止 Telegram 消息接收服务，并返回 polling/typing owner 是否收敛。 + 停止 Telegram 消息接收服务，并返回 SDK/polling/typing owner 是否收敛。 """ converged = True with self._typing_lifecycle_lock: @@ -1757,16 +1773,24 @@ class Telegram: bot = self._bot polling_thread = self._polling_thread + deadline = time.monotonic() + self._shutdown_timeout_seconds + transport_converged = True if bot: - bot.stop_bot() + if not self._stop_bot_with_deadline(bot, deadline): + converged = False + transport_converged = False + logger.error("Telegram SDK worker 未在关闭预算内退出") if ( polling_thread and polling_thread.is_alive() and polling_thread is not threading.current_thread() ): - polling_thread.join(timeout=self._polling_join_timeout_seconds) + polling_thread.join(timeout=max(0.0, deadline - time.monotonic())) if polling_thread and polling_thread.is_alive(): logger.error("Telegram polling 线程未在关闭预算内退出") + converged = False + transport_converged = False + if not transport_converged: return False self._polling_thread = None self._bot = None diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index 615369894..04b39120c 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -27,6 +27,7 @@ from app.runtime.extensions.module_manager import ModuleManager from app.runtime.extensions.module.dispatcher import ModuleInvocationDispatcher from app.runtime.extensions.plugin_manager import PluginManager from app.runtime.events import EventHandlerBinding, EventManager +from app.runtime.execution import run_in_threadpool_to_completion from app.runtime.observability import record_metric from app.runtime.state import SystemHelper from app.runtime.settings import configure_runtime_setting_provider @@ -611,12 +612,16 @@ async def stop_modules() -> bool: name: str, callback: Callable[[], object], *, + offload: bool = False, record_failure: bool = True, ) -> bool: """执行单个关闭步骤，失败时继续收口并保留诚实结果。""" nonlocal all_converged try: - result = callback() + if offload: + result = await run_in_threadpool_to_completion(callback) + else: + result = callback() if inspect.isawaitable(result): result = await result converged = result is not False @@ -633,14 +638,14 @@ async def stop_modules() -> bool: return converged await run_step("图片代理安全日志合并器", close_image_proxy_block_log_coalescer) - await run_step("模块", lambda: ModuleManager().shutdown()) + await run_step("模块", lambda: ModuleManager().shutdown(), offload=True) await run_step("事件消费", lambda: EventManager().stop_async()) - await run_step("浏览器会话", close_browser_sessions) + await run_step("浏览器会话", close_browser_sessions, offload=True) await run_step("托管资源", stop_managed_resources) - await run_step("DoH服务", lambda: DohHelper().shutdown()) - await run_step("线程池", lambda: ThreadHelper().shutdown()) - await run_step("消息服务", stop_message) - await run_step("Redis缓存连接", lambda: RedisHelper().close()) + await run_step("DoH服务", lambda: DohHelper().shutdown(), offload=True) + await run_step("线程池", lambda: ThreadHelper().shutdown(), offload=True) + await run_step("消息服务", stop_message, offload=True) + await run_step("Redis缓存连接", lambda: RedisHelper().close(), offload=True) await run_step("异步Redis缓存连接", lambda: AsyncRedisHelper().close()) # Web Agent 的取消 finally 可能还要写入最终展示快照，必须先完成任务收尾，再关闭写入准入。 web_agent_drained = await run_step( @@ -672,8 +677,8 @@ async def stop_modules() -> bool: else: all_converged = False logger.error("数据库任务未收敛，跳过数据库连接关闭以避免运行中事务使用已释放连接") - await run_step("前端服务", stop_frontend) - await run_step("临时文件", clear_temp) + await run_step("前端服务", stop_frontend, offload=True) + await run_step("临时文件", clear_temp, offload=True) return all_converged diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index a3ad0897e..382e99b55 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -4,7 +4,7 @@ import asyncio import inspect import time from contextlib import asynccontextmanager -from typing import Callable +from typing import Awaitable, Callable from fastapi import FastAPI @@ -33,6 +33,7 @@ from app.foundation.environment import is_free_threaded_runtime, is_gil_enabled settings = RuntimeSettingsCompat() from app.runtime.health import get_application_health +from app.runtime.execution import run_in_threadpool_to_completion from app.runtime.topology import validate_process_topology from app.runtime.tasks import TaskRegistry, configure_task_registry from app.adapters.external.server import MoviePilotServerHelper @@ -135,32 +136,38 @@ async def run_shutdown_step( timeout_seconds: float | None = None, ) -> bool: """在有限预算内执行关闭阶段，并返回资源 owner 是否已经收敛。""" - try: + + async def invoke() -> object: + """在主循环调用 owner，并等待其可能返回的异步结果。""" result = callback() if inspect.isawaitable(result): - task = asyncio.ensure_future(result) + return await result + return result - def _consume_shutdown_result(done: asyncio.Future) -> None: - """消费延迟收敛任务的最终异常，避免事件循环产生未取回异常。""" - try: - done.result() - except asyncio.CancelledError: - pass - except Exception as err: - logger.error(f"关闭{name}最终收尾失败：{err}") + try: + task = asyncio.create_task(invoke(), name=f"shutdown.{name}") - task.add_done_callback(_consume_shutdown_result) - if timeout_seconds: - try: - result = await asyncio.wait_for( - asyncio.shield(task), timeout=timeout_seconds - ) - except asyncio.TimeoutError: - logger.error("关闭%s超时，已请求取消并保留未收敛任务", name) - task.cancel() - return False - else: - result = await task + def _consume_shutdown_result(done: asyncio.Future) -> None: + """消费延迟收敛任务的最终异常，避免事件循环产生未取回异常。""" + try: + done.result() + except asyncio.CancelledError: + pass + except Exception as err: + logger.error(f"关闭{name}最终收尾失败：{err}") + + task.add_done_callback(_consume_shutdown_result) + if timeout_seconds: + try: + result = await asyncio.wait_for( + asyncio.shield(task), timeout=timeout_seconds + ) + except asyncio.TimeoutError: + logger.error("关闭%s超时，已请求取消并保留未收敛任务", name) + task.cancel() + return False + else: + result = await task if result is False: logger.error("关闭%s未收敛，资源所有权保持不变", name) return False @@ -170,6 +177,17 @@ async def run_shutdown_step( return False +def offload_shutdown_callback( + callback: Callable[[], object], +) -> Callable[[], Awaitable[object]]: + """把明确会阻塞的同步关闭 owner 包装为异步生命周期回调。""" + + async def invoke() -> object: + return await run_in_threadpool_to_completion(callback) + + return invoke + + async def run_startup_step( name: str, callback: Callable[[], object], @@ -384,7 +402,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]: dependencies=("插件备份恢复",), mode=LifecycleMode.NORMAL_ONLY, start=init_plugins, - stop=finalize_plugins, + stop=offload_shutdown_callback(finalize_plugins), start_order=90, stop_order=60, start_timeout_seconds=300, @@ -395,7 +413,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]: name="插件变更监控", dependencies=("插件",), mode=LifecycleMode.NORMAL_ONLY, - stop=stop_plugin_monitor, + stop=offload_shutdown_callback(stop_plugin_monitor), stop_order=8, stop_timeout_seconds=10, stop_failure=LifecycleFailurePolicy.FAIL_FAST, @@ -405,7 +423,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]: dependencies=("插件",), mode=LifecycleMode.NORMAL_ONLY, start=init_scheduler, - stop=stop_scheduler, + stop=offload_shutdown_callback(stop_scheduler), start_order=100, stop_order=50, start_timeout_seconds=120, @@ -496,7 +514,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]: dependencies=("命令服务",), mode=LifecycleMode.NORMAL_ONLY, start=init_workflow, - stop=stop_workflow, + stop=offload_shutdown_callback(stop_workflow), start_order=140, stop_order=20, start_timeout_seconds=120, @@ -507,7 +525,9 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]: name="插件备份", dependencies=("插件",), mode=LifecycleMode.NORMAL_ONLY, - stop=lambda: SystemChain().backup_plugins(), + stop=offload_shutdown_callback( + lambda: SystemChain().backup_plugins() + ), stop_order=10, stop_timeout_seconds=300, ), diff --git a/scripts/perf/README.md b/scripts/perf/README.md index bc8775ad8..8fa0421e7 100644 --- a/scripts/perf/README.md +++ b/scripts/perf/README.md @@ -224,6 +224,22 @@ preflight 会验证 Python 3.14、GIL 状态、`thread_inherit_context`、MovieP 依赖、语义、驱动、启动或样本完整性不成立。该工具只用于隔离的本地长 A/B，不接真实凭据、用户数据库、 媒体目录或外网，也不加入常规 CI。 +## 模块关闭事件循环 A/B + +`module_shutdown_ab.py` 使用隔离资源 owner，分别测量 `stop_modules()` 内部同步关闭和生命周期总入口的 +同步关闭。默认制造 50ms 同步等待，并观察 10ms 心跳是否在关闭完成前执行： + +```bash +../.venv-test/bin/python scripts/perf/module_shutdown_ab.py \ + --block-ms 50 \ + --heartbeat-ms 10 \ + --samples 7 +``` + +分别在 Before/After revision 运行相同参数并保存 JSON。总关闭耗时应保持接近固定等待时间；心跳中位延迟 +用于判断事件循环是否被同步 owner 占用，所有样本在关闭完成前执行心跳属于正确性门禁。探针不启动真实 +模块、线程池、数据库、配置或网络。 + ## TaskRegistry 跨线程提交 A/B `task_registry_ab.py` 验证目标事件循环尚未分发 callback 时执行 shutdown，pending completion 与原始 diff --git a/scripts/perf/module_shutdown_ab.py b/scripts/perf/module_shutdown_ab.py new file mode 100644 index 000000000..ca74a924e --- /dev/null +++ b/scripts/perf/module_shutdown_ab.py @@ -0,0 +1,179 @@ +"""测量同步资源关闭期间的事件循环响应与关闭总耗时。""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import statistics +import sys +import time +from pathlib import Path +from types import SimpleNamespace +from typing import Awaitable, Callable + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from app.testing.bootstrap import install_sites_stub, isolate_config_dir + +isolate_config_dir() +install_sites_stub() + +from app.startup import lifecycle +from app.startup.initializers import modules as modules_initializer + + +async def _noop_async() -> None: + """提供不产生外部副作用的异步关闭替身。""" + + +def _noop_sync() -> None: + """提供不产生外部副作用的同步关闭替身。""" + + +def configure_module_probe(block_seconds: float) -> None: + """把模块关闭依赖替换为隔离 owner，仅保留一个固定同步等待。""" + + def blocking_module_shutdown() -> bool: + time.sleep(block_seconds) + return True + + modules_initializer.ModuleManager = lambda: SimpleNamespace( + shutdown=blocking_module_shutdown + ) + modules_initializer.EventManager = lambda: SimpleNamespace( + stop_async=_noop_async + ) + modules_initializer.DohHelper = lambda: SimpleNamespace(shutdown=_noop_sync) + modules_initializer.ThreadHelper = lambda: SimpleNamespace(shutdown=_noop_sync) + modules_initializer.RedisHelper = lambda: SimpleNamespace(close=_noop_sync) + modules_initializer.AsyncRedisHelper = lambda: SimpleNamespace(close=_noop_async) + modules_initializer.close_image_proxy_block_log_coalescer = _noop_async + modules_initializer.close_browser_sessions = _noop_sync + modules_initializer.stop_managed_resources = _noop_async + modules_initializer.stop_message = _noop_sync + modules_initializer.shutdown_web_agent_background_tasks = _noop_async + modules_initializer.wait_web_agent_background_tasks = _noop_async + modules_initializer.get_configured_agent_chat_persistence = lambda: SimpleNamespace( + begin_shutdown=_noop_sync, + shutdown=_noop_async, + ) + + async def stop_database_worker() -> None: + modules_initializer._database_worker = None + + modules_initializer.stop_database_worker = stop_database_worker + modules_initializer.close_database = _noop_async + modules_initializer.stop_frontend = _noop_sync + modules_initializer.clear_temp = _noop_sync + modules_initializer._database_worker = None + + +async def sample( + shutdown: Callable[[], Awaitable[object]], + *, + heartbeat_seconds: float, +) -> dict[str, float | bool]: + """执行一次关闭并测量同期心跳的实际唤醒延迟。""" + loop = asyncio.get_running_loop() + started_at = loop.time() + heartbeat_at: float | None = None + + async def heartbeat() -> None: + nonlocal heartbeat_at + await asyncio.sleep(heartbeat_seconds) + heartbeat_at = loop.time() + + heartbeat_task = asyncio.create_task(heartbeat()) + await shutdown() + shutdown_finished_at = loop.time() + completed_before_shutdown = heartbeat_task.done() + await heartbeat_task + assert heartbeat_at is not None + return { + "heartbeat_delay_ms": (heartbeat_at - started_at) * 1000, + "heartbeat_completed_before_shutdown": completed_before_shutdown, + "shutdown_ms": (shutdown_finished_at - started_at) * 1000, + } + + +async def run_samples( + *, + block_seconds: float, + heartbeat_seconds: float, + samples: int, +) -> dict[str, object]: + """顺序采集模块内部和生命周期总入口两类同步关闭样本。""" + configure_module_probe(block_seconds) + module_samples = [ + await sample( + modules_initializer.stop_modules, + heartbeat_seconds=heartbeat_seconds, + ) + for _ in range(samples) + ] + lifecycle_samples = [ + await sample( + lambda: lifecycle.run_shutdown_step( + "probe.sync_owner", + lifecycle.offload_shutdown_callback( + lambda: time.sleep(block_seconds) + ), + timeout_seconds=max(1.0, block_seconds * 4), + ), + heartbeat_seconds=heartbeat_seconds, + ) + for _ in range(samples) + ] + + def summarize(values: list[dict[str, float | bool]]) -> dict[str, object]: + return { + "samples": values, + "heartbeat_delay_median_ms": statistics.median( + float(value["heartbeat_delay_ms"]) for value in values + ), + "shutdown_median_ms": statistics.median( + float(value["shutdown_ms"]) for value in values + ), + "heartbeat_completed_before_shutdown": all( + bool(value["heartbeat_completed_before_shutdown"]) + for value in values + ), + } + + return { + "block_ms": block_seconds * 1000, + "heartbeat_target_ms": heartbeat_seconds * 1000, + "module_step": summarize(module_samples), + "lifecycle_step": summarize(lifecycle_samples), + } + + +def parse_args() -> argparse.Namespace: + """解析固定阻塞、心跳间隔和样本数。""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--block-ms", type=float, default=50.0) + parser.add_argument("--heartbeat-ms", type=float, default=10.0) + parser.add_argument("--samples", type=int, default=7) + return parser.parse_args() + + +def main() -> None: + """运行隔离样本并输出 JSON。""" + args = parse_args() + if args.block_ms <= 0 or args.heartbeat_ms <= 0 or args.samples < 1: + raise SystemExit("block-ms、heartbeat-ms 和 samples 必须大于 0") + result = asyncio.run( + run_samples( + block_seconds=args.block_ms / 1000, + heartbeat_seconds=args.heartbeat_ms / 1000, + samples=args.samples, + ) + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 480c65a3b..fd88cea46 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6601, - "edge_sha256": "4ebc1db325d75ac04418e89c199dc2b09eb95905e6e03a4101b7780d6b184c7c", + "edge_count": 6603, + "edge_sha256": "70ed941e5ae3893b29167aa191357df42847de4547ce4a0b8d29acf38bbf924d", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -6302,6 +6302,7 @@ "app.startup.initializers.modules -> app.runtime.cache", "app.startup.initializers.modules -> app.runtime.config", "app.startup.initializers.modules -> app.runtime.events", + "app.startup.initializers.modules -> app.runtime.execution", "app.startup.initializers.modules -> app.runtime.extensions", "app.startup.initializers.modules -> app.runtime.extensions.module", "app.startup.initializers.modules -> app.runtime.extensions.module.dispatcher", @@ -6404,6 +6405,7 @@ "app.startup.lifecycle -> app.foundation.environment", "app.startup.lifecycle -> app.runtime", "app.startup.lifecycle -> app.runtime.config", + "app.startup.lifecycle -> app.runtime.execution", "app.startup.lifecycle -> app.runtime.health", "app.startup.lifecycle -> app.runtime.log", "app.startup.lifecycle -> app.runtime.settings", diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index d5f02846e..de66cac93 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -1054,6 +1054,26 @@ def test_stop_modules_propagates_doh_nonconvergence(monkeypatch): _assert_completed_once(dependency) +@pytest.mark.asyncio +async def test_stop_modules_keeps_event_loop_responsive_during_sync_owner_wait( + monkeypatch, +): + """同步 owner 的有界等待不得占用主事件循环。""" + dependencies = _patch_module_shutdown_dependencies(monkeypatch) + release = threading.Event() + timer = threading.Timer(0.05, release.set) + dependencies["module"].side_effect = lambda: release.wait(timeout=1.0) + + heartbeat = asyncio.create_task(asyncio.sleep(0.01)) + timer.start() + try: + await modules_initializer.stop_modules() + finally: + timer.join(timeout=1.0) + + assert heartbeat.done() + + def test_stop_modules_drains_web_agent_tasks_before_persistence(monkeypatch): """关闭时先收口 Web Agent，再关闭持久化准入和数据库任务。""" order = [] @@ -1120,7 +1140,8 @@ async def test_shutdown_timeout_does_not_skip_database_worker_cleanup(monkeypatc "get_configured_agent_chat_persistence", MagicMock(return_value=persistence), ) - stop_database_worker = AsyncMock() + database_worker_stopped = asyncio.Event() + stop_database_worker = AsyncMock(side_effect=database_worker_stopped.set) monkeypatch.setattr(modules_initializer, "stop_database_worker", stop_database_worker) monkeypatch.setattr(modules_initializer, "_database_worker", object()) @@ -1135,6 +1156,7 @@ async def test_shutdown_timeout_does_not_skip_database_worker_cleanup(monkeypatc completed = await shutdown assert completed is False + await asyncio.wait_for(database_worker_stopped.wait(), timeout=1.0) stop_database_worker.assert_awaited_once_with() @@ -1177,6 +1199,54 @@ async def test_shutdown_timeout_has_hard_bound_for_nonconverging_cleanup() -> No await asyncio.wait_for(settled.wait(), timeout=0.2) +@pytest.mark.asyncio +async def test_shutdown_step_bounds_sync_owner_without_blocking_event_loop() -> None: + """同步 owner 超时后应及时返回，并继续持有 worker 直至真实终态。""" + started = threading.Event() + release = threading.Event() + settled = threading.Event() + + def blocking_shutdown() -> None: + started.set() + release.wait(timeout=1.0) + settled.set() + + heartbeat = asyncio.create_task(asyncio.sleep(0.01)) + shutdown = asyncio.create_task( + lifecycle.run_shutdown_step( + "同步阻塞 owner", + lifecycle.offload_shutdown_callback(blocking_shutdown), + timeout_seconds=0.02, + ) + ) + assert await asyncio.to_thread(started.wait, 0.2) + started_at = asyncio.get_running_loop().time() + completed = await shutdown + + assert completed is False + assert asyncio.get_running_loop().time() - started_at < 0.2 + assert heartbeat.done() + assert not settled.is_set() + + release.set() + assert await asyncio.to_thread(settled.wait, 0.2) + + +@pytest.mark.asyncio +async def test_shutdown_step_calls_awaitable_wrapper_on_event_loop() -> None: + """普通 callable 可在主循环构造并返回需要等待的异步结果。""" + loop = asyncio.get_running_loop() + + def shutdown_wrapper() -> asyncio.Task[None]: + assert asyncio.get_running_loop() is loop + return loop.create_task(asyncio.sleep(0)) + + assert await lifecycle.run_shutdown_step( + "异步包装 owner", + shutdown_wrapper, + ) is True + + @pytest.mark.asyncio async def test_shutdown_step_reports_explicit_nonconvergence() -> None: """同步和异步 owner 显式返回 False 时都必须向生命周期传播失败。""" diff --git a/tests/test_module_lifecycle.py b/tests/test_module_lifecycle.py index 0068397bc..491b6be3c 100644 --- a/tests/test_module_lifecycle.py +++ b/tests/test_module_lifecycle.py @@ -1,7 +1,10 @@ +import asyncio import threading +import time from unittest.mock import Mock, patch import pytest +from telebot import TeleBot from app.modules import _MessageBase from app.modules.discord import DiscordModule @@ -23,6 +26,7 @@ from app.modules.wechat import WechatModule from app.modules.wechat.wechatbot import WeChatBot from app.modules.wechatclawbot import WechatClawBotModule from app.modules.wechatclawbot.wechatclawbot import WechatClawBot +from app.runtime.execution import run_in_threadpool_to_completion def test_config_reload_stops_before_initializing_latest_generation(): @@ -180,6 +184,8 @@ def test_telegram_stop_closes_sdk_and_waits_for_polling_thread(): """客户端停止完成后不得保留 SDK worker 或 polling 线程句柄。""" client = Telegram.__new__(Telegram) bot = Mock() + bot.threaded = False + bot.worker_pool = None client._bot = bot polling_thread = Mock() polling_thread.is_alive.side_effect = [True, False] @@ -193,9 +199,9 @@ def test_telegram_stop_closes_sdk_and_waits_for_polling_thread(): assert client.stop() is True assert client.stop() is True - bot.stop_bot.assert_called_once_with() + bot.stop_polling.assert_called_once_with() polling_thread.join.assert_called_once_with( - timeout=client._polling_join_timeout_seconds + timeout=pytest.approx(client._shutdown_timeout_seconds, abs=0.1) ) assert client._bot is None assert client._polling_thread is None @@ -205,11 +211,13 @@ def test_telegram_stop_keeps_polling_owner_when_thread_misses_deadline(): """polling 超过关闭预算时必须返回未收敛并保留原 owner。""" client = Telegram.__new__(Telegram) bot = Mock() + bot.threaded = False + bot.worker_pool = None polling_thread = Mock() polling_thread.is_alive.return_value = True client._bot = bot client._polling_thread = polling_thread - client._polling_join_timeout_seconds = 0.01 + client._shutdown_timeout_seconds = 0.01 client._typing_tasks = {} client._typing_stop_flags = {} client._typing_lock = threading.RLock() @@ -218,11 +226,60 @@ def test_telegram_stop_keeps_polling_owner_when_thread_misses_deadline(): assert client.stop() is False - polling_thread.join.assert_called_once_with(timeout=0.01) + polling_thread.join.assert_called_once() + remaining_timeout = polling_thread.join.call_args.kwargs["timeout"] + assert 0 <= remaining_timeout <= client._shutdown_timeout_seconds assert client._bot is bot assert client._polling_thread is polling_thread +@pytest.mark.asyncio +async def test_telegram_stop_bounds_real_sdk_worker_and_retries_after_release(): + """真实 SDK worker 阻塞时应保留 owner，释放后重试可以完整收敛。""" + bot = TeleBot("123:test", threaded=True, num_threads=1) + entered = threading.Event() + release = threading.Event() + + def blocking_callback() -> None: + entered.set() + release.wait(timeout=1.0) + + bot.worker_pool.put(blocking_callback) + assert await asyncio.to_thread(entered.wait, 0.2) + + client = Telegram.__new__(Telegram) + client._bot = bot + client._polling_thread = None + client._shutdown_timeout_seconds = 0.02 + client._typing_tasks = {} + client._typing_stop_flags = {} + client._typing_lock = threading.RLock() + client._typing_lifecycle_lock = threading.RLock() + client._typing_accepting = True + + heartbeat = asyncio.create_task(asyncio.sleep(0.005)) + started_at = time.monotonic() + try: + assert await run_in_threadpool_to_completion(client.stop) is False + assert time.monotonic() - started_at < 0.2 + assert heartbeat.done() + assert client._bot is bot + assert any(worker.is_alive() for worker in bot.worker_pool.workers) + + release.set() + for worker in bot.worker_pool.workers: + await asyncio.to_thread(worker.join, 0.2) + + assert await run_in_threadpool_to_completion(client.stop) is True + assert client._bot is None + assert client._polling_thread is None + finally: + release.set() + for worker in bot.worker_pool.workers: + worker.stop() + await asyncio.to_thread(worker.join, 0.2) + + @pytest.mark.parametrize( "module_type", [