diff --git a/app/adapters/external/plugin/client.py b/app/adapters/external/plugin/client.py index 7c0a97e0f..1a48473df 100644 --- a/app/adapters/external/plugin/client.py +++ b/app/adapters/external/plugin/client.py @@ -9,9 +9,10 @@ import threading import time import traceback from collections import Counter +from concurrent.futures import Future from pathlib import Path from typing import Any, Dict, Iterator, List, Optional, cast -from urllib.parse import urlparse, urlsplit +from urllib.parse import urlsplit import httpx2 from packaging.specifiers import InvalidSpecifier, SpecifierSet @@ -25,9 +26,11 @@ from app.domain.plugin import ( is_local_plugin_source, is_physical_plugin_id, is_plugin_generation_compatible, + normalize_plugin_market_repo_url as _normalize_plugin_market_repo_url, parse_local_plugin_generation, parse_local_plugin_path, parse_local_plugin_reference, + split_plugin_market_repo_urls as _split_plugin_market_repo_urls, ) from app.foundation.environment import is_free_threaded_runtime from app.foundation.singleton import WeakSingleton @@ -51,6 +54,7 @@ PLUGIN_INDEX_MAX_HISTORY_ENTRIES = 512 PLUGIN_INDEX_MAX_NESTING = 64 PLUGIN_INDEX_READ_CHUNK_SIZE = 64 * 1024 PLUGIN_INDEX_REQUEST_TIMEOUT = 15 +DEFAULT_PLUGIN_INDEX_FETCH_CONCURRENCY = 5 PLUGIN_INDEX_COMPATIBILITY_FLAG_PATTERN = re.compile(r"^v\d+t?$") PLUGIN_INDEX_TEXT_LIMITS = { "name": 256, @@ -64,6 +68,12 @@ PLUGIN_INDEX_TEXT_LIMITS = { "repo_url": 2048, } +# 目录服务和来源库存可能在同一进程内同时刷新；调用方各自创建的信号量 +# 只能限制单次刷新，不能限制真实出站请求总量。索引传输层统一持有闸门。 +_PLUGIN_INDEX_FETCH_GATE = threading.BoundedSemaphore( + DEFAULT_PLUGIN_INDEX_FETCH_CONCURRENCY +) + class _PluginIndexTooLargeError(RuntimeError): """插件索引的声明长度或实际读取字节超过资源边界。""" @@ -139,35 +149,9 @@ VERSION_BACKWARD_COMPATIBLE_FLAGS: Dict[str, List[str]] = { "v3": ["v2"], } -def normalize_plugin_market_repo_url(repo_url: str) -> Optional[str]: - """规范化插件仓库地址，便于跨来源合并去重。""" - repo_url = (repo_url or "").strip().rstrip("/") - if not repo_url: - return None - repo_url = repo_url.removesuffix(".git") - parsed_url = urlparse(repo_url) - if parsed_url.scheme not in {"http", "https"}: - return None - if (parsed_url.hostname or "").lower() != "github.com": - return None - paths = [item for item in parsed_url.path.split("/") if item] - if len(paths) < 2: - return None - return f"https://github.com/{paths[0]}/{paths[1]}" - - -def split_plugin_market_repo_urls(value: Optional[str]) -> list[str]: - """拆分插件市场仓库配置并保持原有顺序去重。""" - repos: list[str] = [] - seen_repos = set() - for item in re.split(r"[\n,，]+", value or ""): - normalized_repo = normalize_plugin_market_repo_url(item) - if not normalized_repo or normalized_repo.lower() in seen_repos: - continue - repos.append(normalized_repo) - seen_repos.add(normalized_repo.lower()) - return repos - +# 保留历史兼容模块的导出路径；规范化规则的唯一实现位于纯领域模块。 +normalize_plugin_market_repo_url = _normalize_plugin_market_repo_url +split_plugin_market_repo_urls = _split_plugin_market_repo_urls def extract_plugin_market_repos_from_wiki( markdown: str, require_markers: bool = False @@ -220,6 +204,13 @@ class PluginMarketTransport(metaclass=WeakSingleton): """负责插件市场、本地仓库和 GitHub 元数据读取。""" _base_url = "https://raw.githubusercontent.com/{user}/{repo}/main/" + _index_future_lock = threading.Lock() + _index_futures: dict[tuple[str, str], Future[Optional[PluginIndex]]] = {} + _index_task_lock = threading.Lock() + _index_tasks: dict[ + tuple[asyncio.AbstractEventLoop, str, str], + asyncio.Task[Optional[PluginIndex]], + ] = {} _release_task_lock = threading.Lock() _release_tasks: dict[ tuple[asyncio.AbstractEventLoop, str, bool], @@ -891,6 +882,8 @@ class PluginMarketTransport(metaclass=WeakSingleton): if not repo_url: return None + repo_url = normalize_plugin_market_repo_url(repo_url) or repo_url + user, repo = cls.get_repo_info(repo_url) if not user or not repo: return None @@ -1137,6 +1130,81 @@ class PluginMarketTransport(metaclass=WeakSingleton): releases.extend(cls.__normalize_plugin_release_response(payload)) return len(payload) >= 100 + @staticmethod + def _plugin_index_key( + repo_url: str, + package_version: Optional[str], + ) -> tuple[str, str]: + """生成跨 URL 表示形式稳定的插件索引请求键。""" + normalized_repo = normalize_plugin_market_repo_url(repo_url) + return ( + (normalized_repo or str(repo_url).strip().rstrip("/")).lower(), + package_version or "", + ) + + @classmethod + def _remove_index_future( + cls, + key: tuple[str, str], + future: Future[Optional[PluginIndex]], + ) -> None: + """同步索引请求结束后释放 single-flight 占位。""" + with cls._index_future_lock: + if cls._index_futures.get(key) is future: + cls._index_futures.pop(key, None) + + @classmethod + def _remove_index_task( + cls, + key: tuple[asyncio.AbstractEventLoop, str, str], + task: asyncio.Task[Optional[PluginIndex]], + ) -> None: + """异步索引请求结束后释放事件循环和仓库引用。""" + with cls._index_task_lock: + if cls._index_tasks.get(key) is task: + cls._index_tasks.pop(key, None) + # 调用方可能在请求完成前取消等待；主动读取异常避免事件循环告警， + # 不影响其他仍在等待同一个任务的调用方取得原始异常。 + try: + task.exception() + except BaseException: # noqa: BLE001 - 仅消费已完成任务的异常 + pass + + @staticmethod + async def _acquire_index_fetch_slot() -> None: + """以可取消方式等待跨同步/异步调用共享的索引请求闸门。""" + while not _PLUGIN_INDEX_FETCH_GATE.acquire(blocking=False): + await asyncio.sleep(0.01) + + def _fetch_plugin_index_sync( + self, + package_url: str, + headers: Optional[dict[str, str]], + ) -> Optional[PluginIndex]: + """在共享进程级闸门内完成一次同步索引读取。""" + with _PLUGIN_INDEX_FETCH_GATE: + response = self.__request_plugin_index_with_fallback( + package_url, + headers=headers, + ) + return self._resolve_plugin_index_result(response) + + async def _fetch_plugin_index_async( + self, + package_url: str, + headers: Optional[dict[str, str]], + ) -> Optional[PluginIndex]: + """在共享进程级闸门内完成一次异步索引读取。""" + await self._acquire_index_fetch_slot() + try: + response = await self.__async_request_plugin_index_with_fallback( + package_url, + headers=headers, + ) + finally: + _PLUGIN_INDEX_FETCH_GATE.release() + return self._resolve_plugin_index_result(response) + @cached(maxsize=1024, ttl=1800, skip_none=False) # type: ignore[misc] def get_plugin_index_result( self, @@ -1148,11 +1216,28 @@ class PluginMarketTransport(metaclass=WeakSingleton): if request is None: raise ValueError("插件仓库地址无效") package_url, headers = request - response = self.__request_plugin_index_with_fallback( - package_url, - headers=headers, - ) - return self._resolve_plugin_index_result(response) + key = self._plugin_index_key(repo_url, package_version) + with self._index_future_lock: + future = self._index_futures.get(key) + owner = future is None + if owner: + future = Future() + self._index_futures[key] = future + + if not owner: + return future.result() + + try: + result = self._fetch_plugin_index_sync(package_url, headers) + except BaseException as error: # noqa: BLE001 - 共享结果必须传播原始异常 + future.set_exception(error) + future.exception() + raise + else: + future.set_result(result) + return result + finally: + self._remove_index_future(key, future) def get_plugins(self, repo_url: str, package_version: Optional[str] = None) -> Optional[PluginIndex]: @@ -1486,11 +1571,32 @@ class PluginMarketTransport(metaclass=WeakSingleton): if request is None: raise ValueError("插件仓库地址无效") package_url, headers = request - response = await self.__async_request_plugin_index_with_fallback( - package_url, - headers=headers, + repo_key, generation_key = self._plugin_index_key( + repo_url, + package_version, ) - return self._resolve_plugin_index_result(response) + loop = asyncio.get_running_loop() + task_key = (loop, repo_key, generation_key) + with self._index_task_lock: + task = self._index_tasks.get(task_key) + if task is None: + task = cast( + asyncio.Task[Optional[PluginIndex]], + get_task_registry().create( + self._fetch_plugin_index_async(package_url, headers), + owner="plugin.market.index", + ), + ) + self._index_tasks[task_key] = task + task.add_done_callback( + lambda completed_task, key=task_key: self._remove_index_task( + key, + completed_task, + ) + ) + + # 单个调用方取消等待时不能连带取消共享请求。 + return await asyncio.shield(task) async def async_get_plugins(self, repo_url: str, package_version: Optional[str] = None) -> Optional[PluginIndex]: diff --git a/app/domain/plugin.py b/app/domain/plugin.py index a749c8070..c6b399943 100644 --- a/app/domain/plugin.py +++ b/app/domain/plugin.py @@ -19,6 +19,39 @@ PLUGIN_GENERATION_COMPATIBILITY: dict[str, tuple[str, ...]] = { } +def normalize_plugin_market_repo_url(repo_url: str) -> str | None: + """规范化 GitHub 插件市场地址，便于配置读取和请求去重。""" + value = str(repo_url or "").strip().rstrip("/") + if not value: + return None + value = value.removesuffix(".git") + try: + parsed_url = urlsplit(value) + except ValueError: + return None + if parsed_url.scheme not in {"http", "https"}: + return None + if (parsed_url.hostname or "").lower() != "github.com": + return None + parts = [item for item in parsed_url.path.split("/") if item] + if len(parts) < 2: + return None + return f"https://github.com/{parts[0]}/{parts[1]}" + + +def split_plugin_market_repo_urls(value: str | None) -> list[str]: + """拆分插件市场配置并按大小写不敏感规则保持顺序去重。""" + repos: list[str] = [] + seen_repos: set[str] = set() + for item in re.split(r"[\n,，]+", value or ""): + normalized_repo = normalize_plugin_market_repo_url(item) + if not normalized_repo or normalized_repo.lower() in seen_repos: + continue + repos.append(normalized_repo) + seen_repos.add(normalized_repo.lower()) + return repos + + @dataclass(frozen=True, slots=True) class PluginReleaseInstallPlan: """描述远端插件内容准备模式，不持有同步或异步 I/O 实现。""" diff --git a/app/runtime/extensions/plugin/catalog.py b/app/runtime/extensions/plugin/catalog.py index 82c58d87a..b54f91a3a 100644 --- a/app/runtime/extensions/plugin/catalog.py +++ b/app/runtime/extensions/plugin/catalog.py @@ -6,6 +6,7 @@ import importlib.util from collections.abc import Callable, Mapping from typing import Any, Optional +from app.domain.plugin import split_plugin_market_repo_urls from app.foundation.version import compare_version from app.runtime.extensions.plugin.contracts import supports_plugin_hook from app.runtime.extensions.plugin.storage import PluginStorage @@ -57,7 +58,7 @@ class PluginCatalogFacade: plugin_market = get_runtime_setting('PLUGIN_MARKET') if not plugin_market: return [] - markets = [item for item in plugin_market.split(",") if item] + markets = split_plugin_market_repo_urls(plugin_market) result = self._market_catalog().collect( markets=markets, compatible_flags=self._system().compatible_flags( @@ -234,7 +235,7 @@ class PluginCatalogFacade: if progress_callback: progress_callback(value=100, text="未配置插件市场，跳过刷新") return [] - markets = [item for item in plugin_market.split(",") if item] + markets = split_plugin_market_repo_urls(plugin_market) result = await self._market_catalog().async_collect( markets=markets, compatible_flags=self._system().compatible_flags( @@ -252,7 +253,7 @@ class PluginCatalogFacade: plugin_market = get_runtime_setting('PLUGIN_MARKET') if not plugin_market: return [] - markets = [item for item in plugin_market.split(",") if item] + markets = split_plugin_market_repo_urls(plugin_market) result: list[Plugin] = await self._market_catalog().async_collect( markets=markets, compatible_flags=self._system().compatible_flags( @@ -280,7 +281,7 @@ class PluginCatalogFacade: def merge(self, higher: list[Plugin], base: list[Plugin]) -> list[Plugin]: """合并不同代际插件目录并保留市场优先级。""" plugin_market = get_runtime_setting('PLUGIN_MARKET') - markets = [item for item in plugin_market.split(",") if item] + markets = split_plugin_market_repo_urls(plugin_market) return self._market_catalog().merge(higher, base, markets) def _safe_state(self, plugin_id: str, plugin: Any) -> bool: diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index bf474bf7d..aaac70335 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -755,7 +755,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 985 | -| 内部导入边 | 8,356 | +| 内部导入边 | 8,358 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 7f14cc72e..933fbf120 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 985 / 8,356 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索与整理任务恢复模块的受控依赖 | +| 宿主 Python 模块 / 内部依赖边 | 985 / 8,358 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索与整理任务恢复模块的受控依赖 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 1b951304e..847a19cf7 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1074,8 +1074,8 @@ "runtime_only": true } }, - "edge_count": 8356, - "edge_sha256": "62b05583fcf296d4f33f9c5ba77eeb8f5a8a200a3809e74ad4df28726587ece4", + "edge_count": 8358, + "edge_sha256": "c0785c61259ea746f28fa0e13d65fb72c3d0d438d52bc624177f0787f98f8728", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -7975,6 +7975,8 @@ "app.runtime.extensions.module.manager -> app.schemas.types", "app.runtime.extensions.plugin.admission -> app.schemas", "app.runtime.extensions.plugin.admission -> app.schemas.exception", + "app.runtime.extensions.plugin.catalog -> app.domain", + "app.runtime.extensions.plugin.catalog -> app.domain.plugin", "app.runtime.extensions.plugin.catalog -> app.foundation", "app.runtime.extensions.plugin.catalog -> app.foundation.version", "app.runtime.extensions.plugin.catalog -> app.runtime", diff --git a/tests/test_plugin_catalog_runtime.py b/tests/test_plugin_catalog_runtime.py index a7f4e3b54..88165024f 100644 --- a/tests/test_plugin_catalog_runtime.py +++ b/tests/test_plugin_catalog_runtime.py @@ -2,11 +2,59 @@ from types import SimpleNamespace +import pytest + from app.runtime.extensions.plugin.catalog import PluginCatalogFacade from app.schemas.plugin import PluginRuntimeStatus from app.schemas.types import SystemConfigKey +@pytest.mark.asyncio +async def test_async_online_normalizes_market_configuration(monkeypatch) -> None: + """目录入口与候选库存使用相同的规范化市场列表。""" + from app.runtime.extensions.plugin import catalog as catalog_module + + requested_markets: list[str] = [] + + class FakeMarketCatalog: + async def async_collect(self, **kwargs): + requested_markets.extend(kwargs["markets"]) + return [] + + monkeypatch.setattr( + catalog_module, + "get_runtime_setting", + lambda key: { + "PLUGIN_MARKET": ( + " https://github.com/Example/Plugins.git/ ," + "https://github.com/example/plugins" + ), + "VERSION_FLAG": "v3", + }[key], + ) + facade = PluginCatalogFacade( + classes=lambda: {}, + running=lambda: {}, + storage=lambda: SimpleNamespace(read=lambda _key: []), + system=lambda: SimpleNamespace( + compatible_flags=lambda _flag: [], + ), + market_catalog=lambda: FakeMarketCatalog(), + market_loader=lambda *_args, **_kwargs: [], + async_market_loader=lambda *_args, **_kwargs: [], + map_plugin=lambda **_kwargs: None, + auth_checker=lambda **_kwargs: True, + plugin_attr=lambda _plugin_id, _attr: None, + plugin_instance=lambda _plugin_id: None, + plugin_instances=lambda: {}, + runtime_status=lambda _plugin_id: None, + log=SimpleNamespace(info=lambda *_args: None), + ) + + assert await facade.async_online() == [] + assert requested_markets == ["https://github.com/Example/Plugins"] + + def test_installed_catalog_keeps_plugins_that_are_not_loaded(): """已安装清单中的插件即使缺依赖或源码也必须保留可观察卡片。""" class ActivePlugin: diff --git a/tests/test_plugin_market_index_policy.py b/tests/test_plugin_market_index_policy.py index 168382210..06129a45a 100644 --- a/tests/test_plugin_market_index_policy.py +++ b/tests/test_plugin_market_index_policy.py @@ -1,10 +1,13 @@ """插件市场索引同步与异步策略一致性测试。""" +import asyncio import json +import threading from contextlib import asynccontextmanager, contextmanager import pytest +from app.adapters.external.plugin import client as plugin_client_module from app.adapters.external.plugin.client import ( PLUGIN_INDEX_MAX_BYTES, PLUGIN_INDEX_MAX_ENTRIES, @@ -13,6 +16,7 @@ from app.adapters.external.plugin.client import ( PluginMarketTransport, _format_request_error, ) +from app.domain.plugin import split_plugin_market_repo_urls SYNC_INDEX_REQUEST = "_PluginMarketTransport__request_plugin_index_with_fallback" ASYNC_INDEX_REQUEST = "_PluginMarketTransport__async_request_plugin_index_with_fallback" @@ -23,6 +27,18 @@ def test_plugin_index_request_timeout_is_bounded() -> None: assert PLUGIN_INDEX_REQUEST_TIMEOUT == 15 +def test_plugin_market_repo_config_is_normalized_once() -> None: + """市场配置的空白、尾斜杠和重复仓库不能制造重复读取任务。""" + assert split_plugin_market_repo_urls( + " https://github.com/Example/Plugins.git/ ," + "https://github.com/example/plugins," + "https://github.com/other/plugins/ " + ) == [ + "https://github.com/Example/Plugins", + "https://github.com/other/plugins", + ] + + def test_github_request_strategies_do_not_retry_direct_when_proxy_is_configured( monkeypatch, ) -> None: @@ -676,6 +692,90 @@ async def test_async_plugin_index_result_preserves_absent_state(monkeypatch) -> assert result is None +@pytest.mark.asyncio +async def test_async_plugin_index_requests_are_single_flight(monkeypatch) -> None: + """同一仓库和代际的并发索引读取只能产生一次出站请求。""" + helper = PluginMarketTransport() + repo_url = "https://github.com/single-flight-owner/single-flight-repo" + started = asyncio.Event() + release = asyncio.Event() + requests = 0 + + async def request(_url: str, *, headers: dict): + nonlocal requests + requests += 1 + started.set() + await release.wait() + return 200, '{"DemoPlugin": {"version": "1.2.3"}}' + + monkeypatch.setattr(helper, ASYNC_INDEX_REQUEST, request) + await helper.async_get_plugin_index_result.cache_clear() + + first = asyncio.create_task( + helper.async_get_plugin_index_result(repo_url, "v3") + ) + await asyncio.wait_for(started.wait(), timeout=1) + second = asyncio.create_task( + helper.async_get_plugin_index_result( + "https://github.com/single-flight-owner/single-flight-repo/", + "v3", + ) + ) + await asyncio.sleep(0) + release.set() + + assert await asyncio.gather(first, second) == [ + {"DemoPlugin": {"version": "1.2.3"}}, + {"DemoPlugin": {"version": "1.2.3"}}, + ] + assert requests == 1 + + +@pytest.mark.asyncio +async def test_async_plugin_index_requests_share_process_gate(monkeypatch) -> None: + """不同目录刷新入口也必须受同一个进程级索引请求闸门约束。""" + helper = PluginMarketTransport() + monkeypatch.setattr( + plugin_client_module, + "_PLUGIN_INDEX_FETCH_GATE", + threading.BoundedSemaphore(2), + ) + active = 0 + peak = 0 + started = asyncio.Event() + release = asyncio.Event() + + async def request(_url: str, *, headers: dict): + nonlocal active, peak + active += 1 + peak = max(peak, active) + if active == 2: + started.set() + await release.wait() + active -= 1 + return 200, '{"DemoPlugin": {"version": "1.2.3"}}' + + monkeypatch.setattr(helper, ASYNC_INDEX_REQUEST, request) + await helper.async_get_plugin_index_result.cache_clear() + tasks = [ + asyncio.create_task( + helper.async_get_plugin_index_result( + f"https://github.com/process-gate-owner/repository-{index}", + "v3", + ) + ) + for index in range(4) + ] + + await asyncio.wait_for(started.wait(), timeout=1) + assert active == 2 + release.set() + results = await asyncio.gather(*tasks) + + assert peak == 2 + assert len(results) == 4 + + def test_plugin_index_result_propagates_adapter_exception(monkeypatch) -> None: """请求异常必须传播给应用库存统一转换为失败事实。""" helper = PluginMarketTransport()