fix(lifecycle): bound shutdown cancellation

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:

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 = {}