From adbeff95d04bdd69c79518c9abde7841b345586b Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Tue, 25 Aug 2026 06:40:43 +0800 Subject: [PATCH] fix(plugin): serialize dynamic route projection (#6441) --- app/adapters/web/plugin/routes.py | 61 +++++++- app/factory.py | 2 + app/runtime/extensions/plugin_manager.py | 27 +++- app/scheduler.py | 9 ++ app/startup/initializers/plugins.py | 2 + .../architecture/dependency-baseline.json | 6 +- tests/test_api_response.py | 143 ++++++++++++++++++ tests/test_plugin_monitor_lifecycle.py | 30 ++++ tests/test_scheduler_cache_expiry.py | 89 +++++++++++ 9 files changed, 365 insertions(+), 4 deletions(-) diff --git a/app/adapters/web/plugin/routes.py b/app/adapters/web/plugin/routes.py index 8bf7515fa..5ebb1c5c8 100644 --- a/app/adapters/web/plugin/routes.py +++ b/app/adapters/web/plugin/routes.py @@ -1,5 +1,8 @@ """FastAPI 动态插件路由适配器。""" +import asyncio +from concurrent.futures import Future, TimeoutError as FutureTimeoutError +from threading import Lock from typing import Any, Callable, Optional from fastapi import Depends, FastAPI @@ -9,6 +12,8 @@ from fastapi.routing import APIRoute class FastAPIDynamicRouteRegistry: """在 FastAPI 上注册插件自由响应路由，并维护 OpenAPI 缓存。""" + _dispatch_admission_timeout = 5.0 + def __init__( self, app: FastAPI, @@ -19,6 +24,7 @@ class FastAPIDynamicRouteRegistry: prefix: str, protected_routes: set[str], log: Any, + event_loop: Callable[[], asyncio.AbstractEventLoop | None] | None = None, ) -> None: """注入应用、插件投影、认证依赖和日志端口。""" self._app = app @@ -29,9 +35,62 @@ class FastAPIDynamicRouteRegistry: self._prefix = prefix self._protected_routes = protected_routes self._logger = log + self._event_loop = event_loop def update(self, plugin_id: Optional[str], action: str) -> None: - """按插件生命周期新增或移除动态路由。""" + """在主事件循环中按插件生命周期新增或移除动态路由。""" + if self._event_loop is None: + self._update(plugin_id, action) + return + target_loop = self._event_loop() + if ( + target_loop is None + or not target_loop.is_running() + or target_loop.is_closed() + ): + raise RuntimeError("主事件循环未运行，无法更新插件动态路由") + try: + current_loop = asyncio.get_running_loop() + except RuntimeError: + current_loop = None + if current_loop is target_loop: + self._update(plugin_id, action) + return + + completed: Future[None] = Future() + dispatch_lock = Lock() + dispatch_started = False + dispatch_abandoned = False + + def apply_update() -> None: + """在目标 loop 的单个回调中完成路由表与 OpenAPI 投影切换。""" + nonlocal dispatch_started + with dispatch_lock: + if dispatch_abandoned: + return + dispatch_started = True + try: + self._update(plugin_id, action) + except BaseException as error: + completed.set_exception(error) + else: + completed.set_result(None) + + target_loop.call_soon_threadsafe(apply_update) + try: + completed.result(timeout=self._dispatch_admission_timeout) + except FutureTimeoutError as error: + with dispatch_lock: + if not dispatch_started: + dispatch_abandoned = True + raise RuntimeError( + "主事件循环未及时接收插件动态路由更新" + ) from error + # 回调一旦开始便不可撤销，等待确定终态以免失败回滚后发生迟到写入。 + completed.result() + + def _update(self, plugin_id: Optional[str], action: str) -> None: + """执行不可中断的路由表与 OpenAPI 投影更新。""" if action not in {"add", "remove"}: raise ValueError("Action must be 'add' or 'remove'") diff --git a/app/factory.py b/app/factory.py index c17da72c5..138622513 100644 --- a/app/factory.py +++ b/app/factory.py @@ -24,6 +24,7 @@ from app.adapters.web.security.access import ( verify_token, ) from app.application.security.token import create_access_token, decode_access_token +from app.runtime.config import global_vars from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() @@ -393,6 +394,7 @@ def create_app() -> FastAPI: "/redoc", }, log=logger, + event_loop=lambda: global_vars.CURRENT_EVENT_LOOP, )) return _app diff --git a/app/runtime/extensions/plugin_manager.py b/app/runtime/extensions/plugin_manager.py index 2726171e4..9c92bae9e 100644 --- a/app/runtime/extensions/plugin_manager.py +++ b/app/runtime/extensions/plugin_manager.py @@ -71,6 +71,7 @@ LegacyPluginImportPreparer = Callable[..., None] PluginInstallReporter = Callable[..., None] SiteAuthLevelProvider = Callable[[], int] PluginCatalogFactory = Callable[["PluginManager"], Any] +PluginRouteRefresher = Callable[[str], None] def _ignore_legacy_diagnostics(**_kwargs) -> None: @@ -109,6 +110,11 @@ def _warn_if_plugin_enabled_gil( ) +def _unavailable_plugin_route_refresher(_plugin_id: str) -> None: + """在 HTTP 组合尚未装配时拒绝发布不完整的热重载投影。""" + raise RuntimeError("插件动态路由刷新器尚未由启动组合根装配") + + _legacy_diagnostics_configurator: LegacyDiagnosticsConfigurator = ( _ignore_legacy_diagnostics ) @@ -119,6 +125,7 @@ _legacy_plugin_import_preparer: LegacyPluginImportPreparer = ( _plugin_install_reporter: PluginInstallReporter = _ignore_legacy_diagnostics _site_auth_level_provider: SiteAuthLevelProvider = _unavailable_site_auth_level _plugin_catalog_factory: PluginCatalogFactory = _unavailable_plugin_catalog_factory +_plugin_route_refresher: PluginRouteRefresher = _unavailable_plugin_route_refresher def configure_plugin_legacy_import_services( @@ -158,6 +165,12 @@ def configure_plugin_catalog_factory(factory: PluginCatalogFactory) -> None: _plugin_catalog_factory = factory +def configure_plugin_route_refresher(refresher: PluginRouteRefresher) -> None: + """由启动组合根注入热重载后的动态路由投影刷新器。""" + global _plugin_route_refresher + _plugin_route_refresher = refresher + + @observe_compat_facade("PluginManager") class PluginManager(ConfigReloadMixin, metaclass=Singleton): """插件管理器""" @@ -656,7 +669,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): monitor_suppressed=self.is_plugin_monitor_suppressed, local_candidate=self._get_local_plugin_candidate_from_path, sync_local=self._sync_local_plugin_if_installed, - reload_plugin=self.reload_plugin_tree, + reload_plugin=self._reload_plugin_tree_from_monitor, dependency_manifest_status=( get_plugin_system().dependency_manifest_status ), @@ -664,6 +677,18 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): log=logger, ).run() + def _reload_plugin_tree_from_monitor( + self, + plugin_id: str, + ) -> PluginRuntimeStatus: + """重载源码树，并发布源插件及虚拟实例的动态路由投影。""" + with self.mutation("热重载插件路由"): + status = self.reload_plugin_tree(plugin_id) + reload_targets = self.get_plugin_reload_targets(plugin_id) + for reload_target in reload_targets: + _plugin_route_refresher(reload_target) + return status + def _get_federated_plugin_change( self, event_path: Path, diff --git a/app/scheduler.py b/app/scheduler.py index 601403839..81d919d21 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -32,6 +32,7 @@ from app.application.agentdata import get_agent_task_port from app.application.database import get_database_governance from app.application.outbox import dispatch_pending_outbox from app.application.plugin.runtime import get_plugin_manager +from app.application.plugin.routes import register_plugin_api from app.application.configuration import ( SchedulerRuntimeConfig, get_configured_system_config, @@ -152,6 +153,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): self._auth_count = 0 # 用户认证失败消息发送 self._auth_message = False + # 插件已按认证结果重建，但动态路由尚未完成投影时保留重试状态。 + self._auth_plugin_routes_pending = False async def on_config_changed(self) -> None: """ @@ -2054,6 +2057,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): """ config = get_scheduler_runtime_config() if SitesHelper().auth_level >= 2: + if self._auth_plugin_routes_pending: + register_plugin_api() + self._auth_plugin_routes_pending = False return # 最大重试次数 __max_try__ = 30 @@ -2088,6 +2094,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): # 认证通过后重新初始化插件 get_plugin_manager().init_config() self.init_plugin_jobs() + self._auth_plugin_routes_pending = True + register_plugin_api() + self._auth_plugin_routes_pending = False else: self._auth_count += 1 diff --git a/app/startup/initializers/plugins.py b/app/startup/initializers/plugins.py index 83a29c234..55942f2a7 100644 --- a/app/startup/initializers/plugins.py +++ b/app/startup/initializers/plugins.py @@ -15,6 +15,7 @@ from app.runtime.extensions.plugin_manager import ( configure_plugin_catalog_factory, configure_plugin_install_reporter, configure_plugin_legacy_import_services, + configure_plugin_route_refresher, configure_plugin_resource_import_preparer, configure_site_auth_level_provider, ) @@ -95,6 +96,7 @@ def configure_plugin_services() -> None: lambda: get_configured_system_config().get(SystemConfigKey.UserInstalledPlugins) or [] ) configure_plugin_catalog_factory(_build_plugin_catalog) + configure_plugin_route_refresher(register_plugin_api) configure_plugin_system(PluginSystemServices( market=market_client, package=PluginPackageManager(plugin_helper), diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 6dc6d0fa8..480c65a3b 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": 6599, - "edge_sha256": "33b44ea7d1f05a4c76fd22c4c1b798689ac8e580099c50141fbd91345a625714", + "edge_count": 6601, + "edge_sha256": "4ebc1db325d75ac04418e89c199dc2b09eb95905e6e03a4101b7780d6b184c7c", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3937,6 +3937,7 @@ "app.factory -> app.application.security", "app.factory -> app.application.security.token", "app.factory -> app.runtime", + "app.factory -> app.runtime.config", "app.factory -> app.runtime.correlation", "app.factory -> app.runtime.localization", "app.factory -> app.runtime.log", @@ -5857,6 +5858,7 @@ "app.scheduler -> app.application.messaging.message", "app.scheduler -> app.application.outbox", "app.scheduler -> app.application.plugin", + "app.scheduler -> app.application.plugin.routes", "app.scheduler -> app.application.plugin.runtime", "app.scheduler -> app.application.scheduling", "app.scheduler -> app.application.site", diff --git a/tests/test_api_response.py b/tests/test_api_response.py index dd535fda9..b5e08406b 100644 --- a/tests/test_api_response.py +++ b/tests/test_api_response.py @@ -1,5 +1,10 @@ +import asyncio +import threading +import time +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace from typing import Any +from unittest.mock import MagicMock import httpx import pytest @@ -928,6 +933,144 @@ async def test_plugin_routes_ignore_included_router_wrappers(): assert removed_response.status_code == 404 +async def test_plugin_route_updates_run_on_application_event_loop() -> None: + """线程侧插件变更必须回投主 loop 后再修改 FastAPI 路由表。""" + app = FastAPI() + main_thread = threading.get_ident() + mutation_threads: list[int] = [] + original_add_api_route = app.router.add_api_route + original_setup = app.setup + + def record_add_api_route(*args, **kwargs): + mutation_threads.append(threading.get_ident()) + return original_add_api_route(*args, **kwargs) + + def record_setup() -> None: + mutation_threads.append(threading.get_ident()) + original_setup() + + app.router.add_api_route = record_add_api_route + app.setup = record_setup + loop = asyncio.get_running_loop() + registry = FastAPIDynamicRouteRegistry( + app=app, + plugin_ids=lambda: ["DemoPlugin"], + plugin_apis=lambda _plugin_id: [{ + "path": "/DemoPlugin/health", + "endpoint": lambda: {"ok": True}, + "methods": ["GET"], + "allow_anonymous": True, + }], + verify_token=lambda: None, + verify_apikey=lambda: None, + prefix="/api/v1/plugin", + protected_routes=set(), + log=SimpleNamespace(debug=lambda *_args: None, error=lambda *_args: None), + event_loop=lambda: loop, + ) + + await asyncio.to_thread(registry.update, "DemoPlugin", "add") + + assert mutation_threads + assert set(mutation_threads) == {main_thread} + assert any( + getattr(route, "path", None) == "/api/v1/plugin/DemoPlugin/health" + for route in app.routes + ) + + +def test_plugin_route_update_rejects_stopped_application_loop() -> None: + """生产 loop 已释放时不得退回调用线程修改路由。""" + registry = FastAPIDynamicRouteRegistry( + app=FastAPI(), + plugin_ids=lambda: [], + plugin_apis=lambda _plugin_id: [], + verify_token=lambda: None, + verify_apikey=lambda: None, + prefix="/api/v1/plugin", + protected_routes=set(), + log=SimpleNamespace(debug=lambda *_args: None, error=lambda *_args: None), + event_loop=lambda: None, + ) + + with pytest.raises(RuntimeError, match="主事件循环未运行"): + registry.update("DemoPlugin", "remove") + + +async def test_plugin_route_update_abandons_late_application_loop_callback() -> None: + """超时前未开始的回调可以迟到执行，但不得再写入路由或污染事件循环。""" + app = FastAPI() + loop = asyncio.get_running_loop() + plugin_apis = MagicMock(return_value=[]) + loop_errors: list[dict] = [] + previous_handler = loop.get_exception_handler() + loop.set_exception_handler(lambda _loop, context: loop_errors.append(context)) + registry = FastAPIDynamicRouteRegistry( + app=app, + plugin_ids=lambda: [], + plugin_apis=plugin_apis, + verify_token=lambda: None, + verify_apikey=lambda: None, + prefix="/api/v1/plugin", + protected_routes=set(), + log=SimpleNamespace(debug=lambda *_args: None, error=lambda *_args: None), + event_loop=lambda: loop, + ) + registry._dispatch_admission_timeout = 0.01 + + try: + with ThreadPoolExecutor(max_workers=1) as executor: + update = executor.submit(registry.update, "DemoPlugin", "add") + time.sleep(0.05) + with pytest.raises(RuntimeError, match="未及时接收"): + update.result() + await asyncio.sleep(0) + finally: + loop.set_exception_handler(previous_handler) + + plugin_apis.assert_not_called() + assert not any( + getattr(route, "path", None) == "/api/v1/plugin/DemoPlugin/health" + for route in app.routes + ) + assert loop_errors == [] + + +async def test_started_plugin_route_update_waits_for_terminal_result() -> None: + """回调开始后即使超过 admission 预算，也不得先报失败再迟到写入。""" + app = FastAPI() + loop = asyncio.get_running_loop() + + def delayed_plugin_apis(_plugin_id: str) -> list[dict]: + time.sleep(0.05) + return [{ + "path": "/DemoPlugin/health", + "endpoint": lambda: {"ok": True}, + "methods": ["GET"], + "allow_anonymous": True, + }] + + registry = FastAPIDynamicRouteRegistry( + app=app, + plugin_ids=lambda: ["DemoPlugin"], + plugin_apis=delayed_plugin_apis, + verify_token=lambda: None, + verify_apikey=lambda: None, + prefix="/api/v1/plugin", + protected_routes=set(), + log=SimpleNamespace(debug=lambda *_args: None, error=lambda *_args: None), + event_loop=lambda: loop, + ) + registry._dispatch_admission_timeout = 0.01 + + await asyncio.to_thread(registry.update, "DemoPlugin", "add") + + assert sum( + getattr(route, "path", None) == "/api/v1/plugin/DemoPlugin/health" + for route in app.routes + ) == 1 + + def test_response_router_uses_response_route_class(): """统一路由器应默认创建统一响应路由。""" router = ResponseAPIRouter() diff --git a/tests/test_plugin_monitor_lifecycle.py b/tests/test_plugin_monitor_lifecycle.py index fb8b10fa8..fa91e5e1e 100644 --- a/tests/test_plugin_monitor_lifecycle.py +++ b/tests/test_plugin_monitor_lifecycle.py @@ -19,6 +19,7 @@ from app.runtime.extensions.plugin.monitor import ( ) from app.runtime.extensions.plugin.admission import PluginMutationAdmission from app.runtime.extensions.plugin.system import reset_plugin_system +from app.runtime.extensions import plugin_manager as plugin_manager_module from app.runtime.extensions.plugin_manager import PluginManager from app.schemas.plugin import PluginRuntimeStatus from app.startup.initializers import plugins as plugins_initializer @@ -749,6 +750,35 @@ def test_plugin_monitor_skips_installing_plugin_until_package_write_finishes(tmp reload_plugin.assert_not_called() +def test_monitor_reload_refreshes_source_and_instance_routes(monkeypatch) -> None: + """源码热重载后必须刷新源插件及其虚拟实例的动态路由投影。""" + admission = PluginMutationAdmission() + manager = SimpleNamespace( + mutation=admission.hold, + get_plugin_reload_targets=MagicMock( + return_value=["DemoPlugin", "DemoPlugin_1"] + ), + reload_plugin_tree=MagicMock(return_value=PluginRuntimeStatus.ACTIVE), + ) + refresh_holds: list[bool] = [] + refresh = MagicMock( + side_effect=lambda _plugin_id: refresh_holds.append(admission.is_held()) + ) + monkeypatch.setattr(plugin_manager_module, "_plugin_route_refresher", refresh) + + status = PluginManager._reload_plugin_tree_from_monitor(manager, "DemoPlugin") + + assert status is PluginRuntimeStatus.ACTIVE + assert admission.active_count == 0 + assert refresh_holds == [True, True] + manager.get_plugin_reload_targets.assert_called_once_with("DemoPlugin") + manager.reload_plugin_tree.assert_called_once_with("DemoPlugin") + assert [item.args for item in refresh.call_args_list] == [ + ("DemoPlugin",), + ("DemoPlugin_1",), + ] + + def test_plugin_monitor_suppression_is_reference_counted(monkeypatch) -> None: """同一插件的重叠写入必须等最后一个事务退出后才解除监控抑制。""" _reset_plugin_manager() diff --git a/tests/test_scheduler_cache_expiry.py b/tests/test_scheduler_cache_expiry.py index 7727acf9f..13ac3c3ed 100644 --- a/tests/test_scheduler_cache_expiry.py +++ b/tests/test_scheduler_cache_expiry.py @@ -2,6 +2,8 @@ import asyncio import threading from unittest.mock import AsyncMock, Mock +import pytest + from app import scheduler as scheduler_module from app.scheduler import Scheduler from app.application.configuration import SchedulerRuntimeConfig @@ -133,3 +135,90 @@ def test_clear_cache_is_manual_only(monkeypatch): assert "clear_cache" in scheduler._jobs assert scheduler._jobs["clear_cache"]["manual"] is True assert background_scheduler.started is True + + +def test_user_auth_refreshes_plugin_routes_after_runtime_reinitialization(monkeypatch): + """自动认证重建插件实例后必须同步刷新动态路由投影。""" + scheduler = object.__new__(Scheduler) + scheduler._auth_count = 1 + scheduler._auth_message = False + scheduler._auth_plugin_routes_pending = False + plugin_manager = Mock() + plugin_jobs = Mock() + refresh_routes = Mock() + message_chain = Mock() + monkeypatch.setattr( + scheduler_module, + "get_scheduler_runtime_config", + lambda: Mock(site_link="https://example.invalid"), + ) + monkeypatch.setattr( + scheduler_module, + "SitesHelper", + lambda: Mock(auth_level=0, check_user=Mock(return_value=(True, "demo"))), + ) + monkeypatch.setattr( + scheduler_module, + "get_configured_system_config", + lambda: Mock(get=Mock(return_value=None)), + ) + monkeypatch.setattr(scheduler_module, "SchedulerChain", lambda: message_chain) + monkeypatch.setattr( + scheduler_module, + "get_plugin_manager", + lambda: plugin_manager, + ) + monkeypatch.setattr(scheduler, "init_plugin_jobs", plugin_jobs) + monkeypatch.setattr(scheduler_module, "register_plugin_api", refresh_routes) + + scheduler.user_auth() + + plugin_manager.init_config.assert_called_once_with() + plugin_jobs.assert_called_once_with() + refresh_routes.assert_called_once_with() + assert scheduler._auth_plugin_routes_pending is False + + +def test_user_auth_retries_pending_plugin_route_projection(monkeypatch): + """认证已成功但路由投影失败时，后续认证任务只重试未完成的投影。""" + scheduler = object.__new__(Scheduler) + scheduler._auth_count = 1 + scheduler._auth_message = False + scheduler._auth_plugin_routes_pending = False + plugin_manager = Mock() + plugin_jobs = Mock() + refresh_routes = Mock(side_effect=[RuntimeError("loop unavailable"), None]) + message_chain = Mock() + sites = Mock(auth_level=0, check_user=Mock(return_value=(True, "demo"))) + monkeypatch.setattr( + scheduler_module, + "get_scheduler_runtime_config", + lambda: Mock(site_link="https://example.invalid"), + ) + monkeypatch.setattr(scheduler_module, "SitesHelper", lambda: sites) + monkeypatch.setattr( + scheduler_module, + "get_configured_system_config", + lambda: Mock(get=Mock(return_value=None)), + ) + monkeypatch.setattr(scheduler_module, "SchedulerChain", lambda: message_chain) + monkeypatch.setattr( + scheduler_module, + "get_plugin_manager", + lambda: plugin_manager, + ) + monkeypatch.setattr(scheduler, "init_plugin_jobs", plugin_jobs) + monkeypatch.setattr(scheduler_module, "register_plugin_api", refresh_routes) + + with pytest.raises(RuntimeError, match="loop unavailable"): + scheduler.user_auth() + + assert scheduler._auth_plugin_routes_pending is True + sites.auth_level = 2 + scheduler.user_auth() + + assert scheduler._auth_plugin_routes_pending is False + assert refresh_routes.call_count == 2 + plugin_manager.init_config.assert_called_once_with() + plugin_jobs.assert_called_once_with() + sites.check_user.assert_called_once_with()