diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index 37fe6bb6b..de01e19a8 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -99,23 +99,30 @@ async def run_shutdown_step( callback: Callable[[], object], timeout_seconds: float | None = None, ) -> None: - """隔离单个关闭阶段的异常,确保后续资源仍有机会释放""" + """在有限预算内执行关闭阶段,并保留未收敛任务的资源所有权。""" try: result = callback() if inspect.isawaitable(result): task = asyncio.ensure_future(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}") + + task.add_done_callback(_consume_shutdown_result) if timeout_seconds: try: await asyncio.wait_for( asyncio.shield(task), timeout=timeout_seconds ) except asyncio.TimeoutError: - logger.error("关闭%s超时,等待其取消收口", name) + logger.error("关闭%s超时,已请求取消并保留未收敛任务", name) task.cancel() - try: - await task - except asyncio.CancelledError: - pass else: await task except Exception as err: diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 6c9996f9f..8f33e084d 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -648,6 +648,44 @@ async def test_shutdown_timeout_does_not_skip_database_worker_cleanup(monkeypatc stop_database_worker.assert_awaited_once_with() +@pytest.mark.asyncio +async def test_shutdown_timeout_has_hard_bound_for_nonconverging_cleanup() -> None: + """关闭收尾不响应取消时,生命周期调用仍必须在预算内返回。""" + started = asyncio.Event() + cancel_requested = asyncio.Event() + release = asyncio.Event() + settled = asyncio.Event() + + async def nonconverging_shutdown() -> None: + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancel_requested.set() + await release.wait() + settled.set() + raise + + started_at = asyncio.get_running_loop().time() + shutdown = asyncio.create_task( + lifecycle.run_shutdown_step( + "不可收敛阶段", + nonconverging_shutdown, + timeout_seconds=0.01, + ) + ) + await started.wait() + await shutdown + + elapsed = asyncio.get_running_loop().time() - started_at + assert elapsed < 0.2 + await asyncio.wait_for(cancel_requested.wait(), timeout=0.2) + assert not settled.is_set() + + release.set() + await asyncio.wait_for(settled.wait(), timeout=0.2) + + def _patch_module_shutdown_dependencies(monkeypatch) -> dict: """替换 stop_modules 的资源所有者,避免测试启动真实后台服务""" dependencies = {}