diff --git a/app/application/plugin/catalog.py b/app/application/plugin/catalog.py index c8a9d5656..366c22159 100644 --- a/app/application/plugin/catalog.py +++ b/app/application/plugin/catalog.py @@ -29,6 +29,7 @@ AsyncMarketLoader = Callable[ ] PluginMapper = Callable[[str, dict, str, list[str], int, Optional[str]], Any] ProgressCallback = Callable[..., Any] +DEFAULT_MARKET_FETCH_CONCURRENCY = 5 @dataclass(frozen=True, slots=True) @@ -103,8 +104,11 @@ class PluginCatalogService: version_compare: Callable[[str, str, str], bool], warning: Callable[[str], Any], error: Callable[[str], Any], + max_concurrency: int = DEFAULT_MARKET_FETCH_CONCURRENCY, ) -> None: """保存市场读取、插件映射和版本比较端口。""" + if max_concurrency < 1: + raise ValueError("插件市场读取并发必须大于 0") self._market_loader = market_loader self._async_market_loader = async_market_loader self._installed_plugins_provider = installed_plugins_provider @@ -113,6 +117,7 @@ class PluginCatalogService: self._version_compare = version_compare self._warning = warning self._error = error + self._max_concurrency = max_concurrency @staticmethod def _load_request( @@ -192,7 +197,9 @@ class PluginCatalogService: loader: Callable[[str, Optional[str], bool], list[Any]], ) -> list[Any]: """并发读取多个市场和代际,并按稳定优先级合并。""" - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor( + max_workers=self._max_concurrency + ) as executor: futures_meta: dict[ concurrent.futures.Future, tuple[int, bool, int], @@ -210,7 +217,11 @@ class PluginCatalogService: collected = [] for future in concurrent.futures.as_completed(futures_meta): - plugins = future.result() + try: + plugins = future.result() + except Exception as error: # noqa: BLE001 - 单个市场失败不能中断全量刷新 + self._error(f"获取插件市场数据失败:{str(error)}") + plugins = [] market_index, is_higher, flag_priority = futures_meta[future] collected.append(( market_index, @@ -240,6 +251,8 @@ class PluginCatalogService: progress_callback: Optional[ProgressCallback] = None, ) -> list[Any]: """异步读取多个市场和代际,并按调用方需要合并或保留仓库候选。""" + semaphore = asyncio.Semaphore(self._max_concurrency) + async def fetch( market: str, package_version: Optional[str], @@ -247,7 +260,8 @@ class PluginCatalogService: task_index: int, ) -> tuple[int, str, list[Any]]: """读取一个市场代际并保留创建时的稳定任务序号。""" - plugins = await loader(market, package_version, force) + async with semaphore: + plugins = await loader(market, package_version, force) return task_index, result_version, plugins or [] tasks: list[asyncio.Task[tuple[int, str, list[Any]]]] = [] diff --git a/tests/test_plugin_catalog_concurrency.py b/tests/test_plugin_catalog_concurrency.py new file mode 100644 index 000000000..c59c3ddf5 --- /dev/null +++ b/tests/test_plugin_catalog_concurrency.py @@ -0,0 +1,114 @@ +"""插件市场目录读取并发控制测试。""" + +import asyncio +import threading +import time +from collections.abc import Callable + +import pytest + +from app.application.plugin.catalog import PluginCatalogService + + +def _catalog( + max_concurrency: int = 2, + error: Callable[[str], None] | None = None, +) -> PluginCatalogService: + """构造不依赖真实插件目录和网络的目录服务。""" + return PluginCatalogService( + market_loader=lambda *_args: {}, + async_market_loader=lambda *_args: {}, + installed_plugins_provider=lambda: [], + plugin_mapper=lambda *_args, **_kwargs: None, + is_local_repo=lambda _repo_url: False, + version_compare=lambda *_args: False, + warning=lambda *_args: None, + error=error if error is not None else lambda *_args: None, + max_concurrency=max_concurrency, + ) + + +def test_sync_collect_limits_market_fetch_concurrency() -> None: + """同步目录刷新不能因市场数量增长而创建无限线程。""" + lock = threading.Lock() + active = 0 + peak = 0 + calls: list[tuple[str, str | None, bool]] = [] + + def loader(market: str, package_version: str | None, force: bool): + nonlocal active, peak + with lock: + active += 1 + peak = max(peak, active) + calls.append((market, package_version, force)) + time.sleep(0.01) + with lock: + active -= 1 + return [] + + result = _catalog(max_concurrency=2).collect( + markets=["market-a", "market-b", "market-c"], + compatible_flags=["v3"], + force=True, + loader=loader, + ) + + assert result == [] + assert len(calls) == 6 + assert peak <= 2 + + +def test_sync_collect_isolates_market_failure() -> None: + """同步目录刷新遇到单个市场异常时仍保留其他市场结果。""" + errors: list[str] = [] + + def loader( + market: str, + package_version: str | None, + _force: bool, + ): + if market == "market-b" and package_version == "v3": + raise RuntimeError("market unavailable") + return [] + + result = _catalog(max_concurrency=2, error=errors.append).collect( + markets=["market-a", "market-b"], + compatible_flags=["v3"], + force=True, + loader=loader, + ) + + assert result == [] + assert len(errors) == 1 + assert "market unavailable" in errors[0] + + +@pytest.mark.asyncio +async def test_async_collect_limits_market_fetch_concurrency() -> None: + """异步目录刷新必须用信号量限制同时在途的市场请求。""" + active = 0 + peak = 0 + + async def loader(_market: str, _package_version: str | None, _force: bool): + nonlocal active, peak + active += 1 + peak = max(peak, active) + await asyncio.sleep(0.01) + active -= 1 + return [] + + result = await _catalog(max_concurrency=2).async_collect( + markets=["market-a", "market-b", "market-c"], + compatible_flags=["v3"], + force=True, + loader=loader, + ) + + assert result == [] + assert peak <= 2 + + +def test_market_fetch_concurrency_must_be_positive() -> None: + """无效并发值应在服务构造时快速失败。""" + with pytest.raises(ValueError, match="大于 0"): + _catalog(max_concurrency=0)