diff --git a/app/runtime/config.py b/app/runtime/config.py index 1cc0ec64c..af0f96e96 100644 --- a/app/runtime/config.py +++ b/app/runtime/config.py @@ -1361,6 +1361,11 @@ class GlobalVar(object): # 生命周期登记的主事件循环 CURRENT_EVENT_LOOP: Optional[AbstractEventLoop] = None + def __init__(self) -> None: + self.CURRENT_EVENT_LOOP = None + self._event_loop_owners: dict[object, AbstractEventLoop] = {} + self._event_loop_owner_lock = threading.Lock() + def stop_system(self): """ 停止系统 @@ -1456,14 +1461,24 @@ class GlobalVar(object): raise RuntimeError("主事件循环尚未启动或已经停止") return loop - def set_loop(self, loop: AbstractEventLoop) -> None: - """登记承载主程序异步任务的事件循环。""" - self.CURRENT_EVENT_LOOP = loop + def set_loop(self, loop: AbstractEventLoop) -> object: + """登记主事件循环，并返回仅供当前生命周期释放的 owner。""" + owner = object() + with self._event_loop_owner_lock: + self._event_loop_owners[owner] = loop + self.CURRENT_EVENT_LOOP = loop + return owner - def clear_loop(self, loop: AbstractEventLoop) -> None: - """仅在登记值仍为目标循环时清除主事件循环。""" - if self.CURRENT_EVENT_LOOP is loop: - self.CURRENT_EVENT_LOOP = None + def clear_loop(self, owner: object) -> None: + """释放指定 owner，保留仍然有效的其他生命周期登记。""" + with self._event_loop_owner_lock: + if owner not in self._event_loop_owners: + return + self._event_loop_owners.pop(owner) + self.CURRENT_EVENT_LOOP = next( + reversed(self._event_loop_owners.values()), + None, + ) # 全局标识 diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index 9e06a4471..ab197a123 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -513,6 +513,7 @@ async def lifespan(app: FastAPI): health = get_application_health(app) health.begin_startup() main_loop = asyncio.get_running_loop() + main_loop_owner: object | None = None enabled_components: tuple[LifecycleComponent, ...] = () started_component_names: set[str] = set() active_start_component: LifecycleComponent | None = None @@ -522,7 +523,7 @@ async def lifespan(app: FastAPI): safe_mode=settings.MOVIEPILOT_SAFE_MODE, ) print("Starting up...") - global_vars.set_loop(main_loop) + main_loop_owner = global_vars.set_loop(main_loop) components = build_lifecycle_components(app) enabled_components = tuple( component @@ -567,7 +568,8 @@ async def lifespan(app: FastAPI): except Exception as cleanup_error: logger.error(f"启动失败后的生命周期清理失败：{cleanup_error}") finally: - global_vars.clear_loop(main_loop) + if main_loop_owner is not None: + global_vars.clear_loop(main_loop_owner) raise try: # 在此处 yield，表示应用已经启动，控制权交回 FastAPI 主事件循环 @@ -585,4 +587,5 @@ async def lifespan(app: FastAPI): # 日志最后关闭，确保其他组件的收尾信息已写入文件 LoggerManager.shutdown() finally: - global_vars.clear_loop(main_loop) + if main_loop_owner is not None: + global_vars.clear_loop(main_loop_owner) diff --git a/tests/test_global_event_loop.py b/tests/test_global_event_loop.py index 208225535..a69503bc5 100644 --- a/tests/test_global_event_loop.py +++ b/tests/test_global_event_loop.py @@ -47,14 +47,33 @@ def test_clear_global_loop_preserves_new_owner() -> None: async def verify() -> None: current = asyncio.get_running_loop() - runtime.set_loop(current) - runtime.clear_loop(previous) + previous_owner = runtime.set_loop(previous) + current_owner = runtime.set_loop(current) + runtime.clear_loop(previous_owner) assert runtime.loop is current - runtime.clear_loop(current) + runtime.clear_loop(current_owner) assert runtime.CURRENT_EVENT_LOOP is None try: asyncio.run(verify()) finally: previous.close() + + +def test_nested_owner_release_restores_same_event_loop() -> None: + """同一循环上的内层生命周期退出后，外层 owner 仍保持登记。""" + runtime = GlobalVar() + + async def verify() -> None: + loop = asyncio.get_running_loop() + outer_owner = runtime.set_loop(loop) + inner_owner = runtime.set_loop(loop) + + runtime.clear_loop(inner_owner) + assert runtime.loop is loop + + runtime.clear_loop(outer_owner) + assert runtime.CURRENT_EVENT_LOOP is None + + asyncio.run(verify()) diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 733cf8966..9b74200a0 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -152,8 +152,9 @@ def test_lifespan_normal_mode_starts_full_runtime(monkeypatch): asyncio.run(run_lifespan()) - configured_loop = lifecycle.global_vars.set_loop.call_args.args[0] - lifecycle.global_vars.clear_loop.assert_called_once_with(configured_loop) + lifecycle.global_vars.clear_loop.assert_called_once_with( + lifecycle.global_vars.set_loop.return_value + ) lifecycle.init_modules.assert_awaited_once_with() lifecycle.prepare_database_component.assert_called_once() lifecycle.configure_plugin_services.assert_called_once_with() @@ -170,6 +171,26 @@ def test_lifespan_normal_mode_starts_full_runtime(monkeypatch): _assert_completed_once(step) +def test_lifespan_validation_failure_does_not_clear_outer_loop_owner(monkeypatch): + """当前生命周期尚未取得 owner 时，启动失败不得清理外层登记。""" + _patch_lifespan(monkeypatch) + monkeypatch.setattr( + lifecycle, + "validate_process_topology", + MagicMock(side_effect=RuntimeError("invalid topology")), + ) + + async def run_lifespan(): + async with lifecycle.lifespan(FastAPI()): + pass + + with pytest.raises(RuntimeError, match="invalid topology"): + asyncio.run(run_lifespan()) + + lifecycle.global_vars.set_loop.assert_not_called() + lifecycle.global_vars.clear_loop.assert_not_called() + + @pytest.mark.parametrize( ("failing_step", "completed_steps", "blocked_steps"), [ @@ -741,8 +762,9 @@ def test_lifespan_fails_fast_when_async_engine_cannot_be_built(monkeypatch): with pytest.raises(RuntimeError, match="no async driver"): asyncio.run(run_lifespan()) - configured_loop = lifecycle.global_vars.set_loop.call_args.args[0] - lifecycle.global_vars.clear_loop.assert_called_once_with(configured_loop) + lifecycle.global_vars.clear_loop.assert_called_once_with( + lifecycle.global_vars.set_loop.return_value + ) # 失败要发生在任何东西被初始化之前，否则模块起来了却没人关：关停块在 yield 处才开始 lifecycle.init_routers.assert_not_called() lifecycle.init_modules.assert_not_called()